基于libusb的mtp通信实现

#include <string>
#include <fstream>
#include <memory>
#include <iostream>
using namespace std;

#include "ptp.h"
#include "ptpcam.h"

#define PTP_DP_NODATA       0x0000  /* no data phase */
#define PTP_DP_SENDDATA     0x0001  /* sending data */
#define PTP_DP_GETDATA      0x0002  /* receiving data */

#define True 1
#define False -1

extern "C" __declspec(dllexport) int __cdecl init_cam(unsigned short, unsigned short);
extern "C" __declspec(dllexport) void __cdecl free_cam();
extern "C" __declspec(dllexport) char* __cdecl send_msg(char*, int N = 10);

extern "C" __declspec(dllexport) int __cdecl open_session();
extern "C" __declspec(dllexport) int __cdecl close_session();
extern "C" __declspec(dllexport) void __cdecl set_buf_size(int);
extern "C" __declspec(dllexport) void __cdecl set_delay(int);
extern "C" __declspec(dllexport) int __cdecl check_usb_connection(int, int);

usb_dev_handle * get_usb_dev(unsigned short, unsigned short);

static struct usb_device * dev;
static usb_dev_handle * hDev;
static PTPParams * params;
static PTP_USB * ptp_usb;

static char * pBuf;
static int buf_size = 512000;
static int msg_delay = 500;
static int file_delay = 100;

#include "mtprequest.h"

int init_cam(unsigned short vid, unsigned short pid)
{
dev = NULL;
hDev = NULL;
params = NULL;
ptp_usb = NULL;

hDev = get_usb_dev(vid, pid);
if (dev && hDev)
{
usb_set_configuration(hDev, dev->config->bConfigurationValue);
if (0 == usb_claim_interface(hDev, dev->config->interface->altsetting->bInterfaceNumber))
{
params = new PTPParams;
ptp_usb = new PTP_USB;
memset(params, 0, sizeof(PTPParams));
memset(ptp_usb, 0, sizeof(PTP_USB));

ptp_usb->handle = hDev;
find_endpoints(dev, &ptp_usb->inep, &ptp_usb->outep, &ptp_usb->intep);
init_ptp_usb(params, ptp_usb, hDev);

pBuf = (char*)malloc(sizeof(char) * buf_size);
memset(pBuf, 0, sizeof(char) * buf_size);
return True;
}
usb_reset(hDev);
usb_close(hDev);
hDev = NULL;
dev = NULL;
}
return False;
}

void free_cam()
{
if (hDev)
{
close_usb(ptp_usb, dev);
dev = NULL;
hDev = NULL;
}
if (params)
{
delete params;
params = NULL;
}
if (ptp_usb)
{
delete ptp_usb;
ptp_usb = NULL;
}
if (pBuf)
{
free(pBuf);
pBuf = NULL;
}
}

usb_dev_handle * get_usb_dev(unsigned short vid, unsigned short pid)
{
dev = NULL;
for (usb_bus * bus = init_usb(); bus; bus = bus->next)
{
for (dev = bus->devices; dev; dev = dev->next)
{
if (vid == dev->descriptor.idVendor && pid == dev->descriptor.idProduct)
{
return usb_open(dev);
}
}
}
dev = NULL;
return NULL;
}

char * send_msg(char* msg, int N)
{
memset(pBuf, 0, buf_size);
PTPContainer * ptp = new PTPContainer;
ptp->Code = code; //please reset the code.
ptp->Param1 = strlen(msg);
ptp->Nparam = 1;
if (PTP_RC_OK == ptp_transaction(params, ptp, PTP_DP_SENDDATA, strlen(msg), &msg))
{
ptp->Code = code; //please reset the code.
ptp->Nparam = 0;
while (N > 0)
{
ptp_transaction(params, ptp, PTP_DP_GETDATA, 0, &pBuf);
N--;
try
{
check_resp(pBuf);
break;
}
catch (RespError)
{
Sleep(msg_delay);
continue;
}
}

}

delete ptp;
return pBuf;
}

int open_session()
{
if (PTP_RC_OK == ptp_opensession(params, 1))
return True;
else
return False;
}

int close_session()
{
if (PTP_RC_OK == ptp_closesession(params))
return True;
else
return False;
}

void set_buf_size(int size)
{
buf_size = size;
}

void set_delay(int d)
{
msg_delay = d;
}

int check_usb_connection(int vid, int pid)
{
struct usb_device * pDev = NULL;
for (usb_bus * bus = init_usb(); bus; bus = bus->next)
{
for (pDev = bus->devices; pDev; pDev = pDev->next)
{
if (vid == pDev->descriptor.idVendor && pid == pDev->descriptor.idProduct)
{
return 1;
}
}
}
return 0;
}

from ctypes import *
import json

class MtpRequest(object):
VID = vid # according to device
PID = pid # according to device
DLL_PATH = "dllname.dll"
hdll = CDLL(DLL_PATH)

def __init__(self, vid=VID, pid=PID, path=DLL_PATH):
if not MtpRequest.hdll:
MtpRequest.hdll = CDLL(path)
self.hdll = MtpRequest.hdll

if c_int(self.hdll.init_cam(vid, pid)).value < 0:
raise RuntimeError("init camera failed")

self.hdll.close_session() #没有这一句有时候open_session会失败

if c_int(self.hdll.open_session()).value < 0:
raise RuntimeError("open session failed")

def __del__(self):
self.hdll.close_session()
self.hdll.free_cam()

def send_msg(self, msg, n=10):
"""
:param msg: dict
:param n: int
:return dict:
"""
resp = c_char_p(self.hdll.send_msg(json.dumps(msg), n)).value
if resp.find("rval") < 0:
raise RuntimeError("MsgError")
return json.loads(resp)