我已连接到 HikVision 摄像头流，并且能够隔离图片数据，但无法将该数据转换为图像。如果我将二进制数据保存到图像文件，该文件不会显示任何图像。如果我使用 Pillow ，它会说未识别图像。
这是图像数据：https://drive.google.com/file/d/1z1TfPb88Ng5qkXWZZ9QfrZR4H5fh43pg/view?usp=sharing ..
对于 stackoverflow 或 Pastebin 来说文本太长..
这就是我从流中获取数据的方式：
with requests.get(url, auth=auth, headers=headers, stream=True) as response:
response_text = ''
license_plate = ''
direction = ''
color = ''
is_picture = False
picture_data = b''
for line in response.iter_lines():
if not line:
continue
line_str = str(line)
if 'Content-Type: image/jpeg' in line_str:
is_picture = True
picture_data = b''
elif is_picture:
picture_data += line
response_text += line.decode('utf-8', errors='ignore')
if '--boundary' not in response_text:
continue
if is_picture:
with open('picture_data.txt', 'wb') as f:
f.write(picture_data)
print(picture_data)
print('picture saved')
is_picture = False
# picture_data = b''
response_text = response_text.rstrip('--boundary')
split_text = response_text.split('<EventNotificationAlert')
if len(split_text) > 1:
notification_alert = '<EventNotificationAlert' + split_text[1]
event_type = '<eventType></eventType>'
event_type = re.search('<eventType>(.*?)</eventType>', notification_alert).group(1)
if event_type == 'ANPR':
with open('test.xml', 'a') as f:
f.writelines(notification_alert)
else:
print('not anpr')
else:
print("'<EventNotificationAlert' not found in response_text")
response_text = ''
这就是我尝试解码图片数据的方式：
bytes = picture_data
a = bytes.find(b'\xff\xd8')
b = bytes.find(b'\xff\xd9')
if a != -1 and b != -1:
jpg = bytes[a:b+8]
bytes = bytes[b+2:]
i = Image.open(BytesIO(jpg))
buffered = BytesIO()
i.save(buffered, format="JPEG")
byte_im = buffered.getvalue()
这不起作用，PIL 库无法识别任何图像。
我可以直接使用以下方式从相机获取图像：
def get_picture_direct(self, registration):
url = f'http://{self.stream_url}/ISAPI/Streaming/channels/101/picture'
auth = HTTPDigestAuth(self.stream_username, self.stream_password)
response = requests.get(url, auth=auth)
jpg = response.content
i = Image.open(BytesIO(jpg))
buffered = BytesIO()
i.save(buffered, format="JPEG")
byte_im = buffered.getvalue()
url=self.upload_image_to_firestore(vehicle_registration=registration, bucket_name=f'{self.storage_bucket}', data=byte_im)
return url
来自摄像机的事件流发送 ANPR 事件通知，并且每个 ANPR 通知附带一张图像。如果我在通知后使用直接方法，图像捕捉得太晚并且错过了车牌号。
如何正确地将字节数据转换为图像？
非常感谢任何帮助。
这是通过python接收anpr报警并保存车牌号和车牌号图像的示例代码：
from http.server import BaseHTTPRequestHandler, HTTPServer
import xml.etree.ElementTree as ET
import re
import csv
import os
class ANPRRequestHandler(BaseHTTPRequestHandler):
def do_POST(self):
# Read the body of the request
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length)
# Split the payload into parts
parts = post_data.split(b'--boundary\r\n')[1:-1] # Exclude the first and last split elements
for part in parts:
headers, body = part.split(b'\r\n\r\n', 1)
body = body.strip()
# Process XML part
if b'anpr.xml' in headers:
self.process_xml(body)
# Process image part
elif b'.jpg' in headers:
filename = re.findall(rb'filename="([^"]+)"', headers)[0].decode()
self.save_image(body, filename)
# Send response
self.send_response(200)
self.end_headers()
self.wfile.write("Received and processed ANPR event".encode('utf-8'))
def append_to_csv(self, license_plate):
# CSV file path
csv_file = 'license_plates.csv'
# Check if the CSV file exists
file_exists = os.path.isfile(csv_file)
# Open the file in append mode
with open(csv_file, mode='a', newline='') as file:
writer = csv.writer(file)
# If the file does not exist, write the header
if not file_exists:
writer.writerow(['License Plate'])
# Write the license plate data
writer.writerow([license_plate])
def process_xml(self, xml_data):
# Parse XML data
#print(xml_data)
try:
root = ET.fromstring(xml_data)
# Find the license plate element
# Update XPath to include namespace prefix
namespaces = {'ns': 'http://www.hikvision.com/ver20/XMLSchema'}
# Find the license plate element
# Update XPath to include namespace prefix
license_plate_element = root.find('.//ns:ANPR/ns:licensePlate', namespaces)
if license_plate_element is not None and license_plate_element.text:
license_plate = license_plate_element.text.strip()
print(f"License Plate: {license_plate}")
self.append_to_csv(license_plate)
else:
print("License plate not found in XML.")
except ET.ParseError as e:
print(f"Error parsing XML: {e}")
def save_image(self, image_data, image_name):
# Save the image
with open(image_name, 'wb') as image_file:
image_file.write(image_data)
print(f"Saved image {image_name}")
def run(server_class=HTTPServer, handler_class=ANPRRequestHandler, port=8888):
server_address = ('', port)
httpd = server_class(server_address, handler_class)
print(f"Starting httpd on port {port}...")
httpd.serve_forever()
if __name__ == '__main__':
run()