This commit is contained in:
yair-mantis 2023-12-29 02:26:24 +02:00
parent e1d688da6e
commit e15ec35ddd
3 changed files with 142 additions and 18 deletions

1
.gitignore vendored
View File

@ -1 +1,2 @@
/venv /venv
/frames

82
cvview.py Normal file
View File

@ -0,0 +1,82 @@
import os
import cv2
import argparse
# Set up the argument parser
parser = argparse.ArgumentParser(description="Visualize image files and display pixel values on hover.")
parser.add_argument('path', help='The path to an image file.')
# Parse the arguments
args = parser.parse_args()
img_path = args.path
# Function to display the image and pixel values along with the frame index
def show_pixel_values(image_path):
def mouse_event(event, x, y, flags, param):
if event == cv2.EVENT_MOUSEMOVE:
pixel_value = img[y, x]
text = f'Value: {pixel_value}, Location: ({x},{y})'
img_text = img.copy()
# Overlay the frame index
frame_index = get_frame_index(image_path)
cv2.putText(img_text, f'Frame: {frame_index}', (10, img_text.shape[0] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 1, cv2.LINE_AA)
cv2.putText(img_text, text, (50, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 1, cv2.LINE_AA)
cv2.imshow('Image', img_text)
img = cv2.imread(image_path, cv2.IMREAD_UNCHANGED)
if img is None:
print(f"Failed to load image at {image_path}. Check the file path and integrity.")
return False
cv2.namedWindow('Image')
cv2.setMouseCallback('Image', mouse_event)
cv2.imshow('Image', img)
return True
# Function to get the frame index from the filename
def get_frame_index(filename):
return os.path.splitext(os.path.basename(filename))[0][-4:]
# Function to modify the numeric part of the filename
def modify_filename(filename, increment=True):
directory, basename = os.path.split(filename)
basename_no_ext, ext = os.path.splitext(basename)
print(f"Modifying filename {basename_no_ext} in directory {directory}.")
if len(basename_no_ext) < 4 or not basename_no_ext[-4:].isdigit():
raise ValueError("Filename does not end with five digits.")
num_part = basename_no_ext[-4:]
num = int(num_part) + (1 if increment else -1)
new_name = f"{basename_no_ext[:-4]}{num:04d}{ext}"
new_path = os.path.join(directory, new_name)
if not os.path.exists(new_path):
print(f"No file found at {new_path}.")
return filename # Return the original filename if the new file does not exist
return new_path
# Ensure the provided path is a valid file
if not os.path.isfile(img_path):
print("The provided path is not a valid file.")
exit(1)
# Initially display the image
if not show_pixel_values(img_path):
exit(1)
# Main loop to navigate through images
while True:
key = cv2.waitKey(0)
if key == 27: # ESC key to exit
break
elif key == 91: # '[' key
img_path = modify_filename(img_path, increment=False)
elif key == 93: # ']' key
img_path = modify_filename(img_path, increment=True)
# Show the new image
if not show_pixel_values(img_path):
break # Exit the loop if the new image cannot be loaded
cv2.destroyAllWindows()

View File

@ -1,4 +1,6 @@
from os import system import argparse
import os
import subprocess
import numpy as np import numpy as np
from tqdm import tqdm from tqdm import tqdm
import pandas as pd import pandas as pd
@ -6,10 +8,24 @@ import pcapng
from struct import unpack from struct import unpack
from PIL import Image from PIL import Image
# Read packets from a pcap file
scanner = pcapng.scanner.FileScanner(open('in.pcap', 'rb'))
blocks = list(tqdm(scanner))
# Create the parser
parser = argparse.ArgumentParser(description="Process a pcap file.")
# Add an argument for the pcap file, with a default value
parser.add_argument('input_file', nargs='?', default='in.pcap', help='The pcap file to process')
# Parse the arguments
args = parser.parse_args()
# Now use args.input_file as the file to process
input_file = args.input_file
basename = os.path.splitext(os.path.basename(input_file))[0]
# Read packets from a pcap file
scanner = pcapng.scanner.FileScanner(open(input_file, "rb"))
blocks = list(tqdm(scanner))
# Helper function to safely get an attribute from an object # Helper function to safely get an attribute from an object
def tryget(obj, att): def tryget(obj, att):
@ -19,31 +35,33 @@ def tryget(obj, att):
# Create a DataFrame with packet lengths # Create a DataFrame with packet lengths
df = pd.DataFrame([{'index': i, 'length': tryget(obj, 'packet_len')} for i, obj in enumerate(blocks)]) df = pd.DataFrame(
[{"index": i, "length": tryget(obj, "packet_len")} for i, obj in enumerate(blocks)]
)
# Filter and extract data packets of a specific length # Filter and extract data packets of a specific length
data = [blocks[i] for i in df[df.length == 6972.0].index] data = [blocks[i] for i in df[df.length == 6972.0].index]
# Remove the UDP header from the packets # Remove the UDP header from the packets
raw = [d.packet_data[0x2a:] for d in data] raw = [d.packet_data[0x2A:] for d in data]
# Function to parse packet data # Function to parse packet data
def parse(data): def parse(data):
hdr = 4 + 2 * 7 # Header length hdr = 4 + 2 * 7 # Header length
# Unpack data into variables # Unpack data into variables
c1, c2, part, a, ffaa, b, c, d = unpack('>Lhhhhhhh', data[:hdr]) c1, c2, part, a, ffaa, b, c, d = unpack(">Lhhhhhhh", data[:hdr])
ret = locals() ret = locals()
del ret['data'] del ret["data"]
del ret['hdr'] del ret["hdr"]
ret['data'] = data[hdr:] ret["data"] = data[hdr:]
return ret return ret
# Parse each packet and create a DataFrame # Parse each packet and create a DataFrame
df = pd.DataFrame([parse(d) for d in raw]) df = pd.DataFrame([parse(d) for d in raw])
df2 = df[[c for c in df.columns if c != 'data']] df2 = df[[c for c in df.columns if c != "data"]]
# Function to group data into frames # Function to group data into frames
@ -55,16 +73,17 @@ def getframes(df):
if len(current) > 0: if len(current) > 0:
frames.append(current) frames.append(current)
current = [] current = []
current.append(row['data']) current.append(row["data"])
if len(current) > 0: if len(current) > 0:
frames.append(current) frames.append(current)
return [b''.join(parts) for parts in frames] return [b"".join(parts) for parts in frames]
# Function to convert binary frame data into images # Function to convert binary frame data into images
def image16(frame, width, height, pixelformat='>H'): def image16(frame, width, height, pixelformat=">H"):
return [ return [
Image.fromarray(np.frombuffer(frame, dtype=pixelformat).reshape(width, height)) for frame in frames Image.fromarray(np.frombuffer(frame, dtype=pixelformat).reshape(width, height))
for frame in frames
if len(frame) == 2 * width * height if len(frame) == 2 * width * height
] ]
@ -73,10 +92,32 @@ def image16(frame, width, height, pixelformat='>H'):
frames = getframes(df) frames = getframes(df)
images = image16(frames, 384, 288) images = image16(frames, 384, 288)
# Create the directory for frames if not exists
frame_dir = f"frames/{basename}"
if not os.path.exists(frame_dir):
os.makedirs(frame_dir)
# Save each image as a PNG file # Save each image as a PNG file
for i, img in enumerate(tqdm(images)): for i, img in enumerate(tqdm(images)):
img.save(f'{i:04}.png') img.save(f'frames/{basename}/{basename}_{i:04}.png')
# Produce a video from the saved images # Produce a video from the saved images
system('ffmpeg -hide_banner -loglevel info -y -f image2 -framerate 25 -i %04d.png -vf "transpose=1" -s 384x288 -vcodec libx264 -pix_fmt yuv420p thermal.mp4') ffmpeg_input = f"frames/{basename}/{basename}_%04d.png"
print('to play: ffplay thermal.mp4') command = [
"ffmpeg",
"-y", # Overwrite output file without asking
"-hide_banner", # Hide banner
"-loglevel", "info", # Log level
"-f", "image2", # Input format
"-framerate", "25", # Framerate
"-i", ffmpeg_input, # Input file pattern
"-vf", "transpose=1", # Video filter for transposing
"-s", "384x288", # Size of one frame
"-vcodec", "libx264", # Video codec
"-pix_fmt", "yuv420p", # Pixel format: YUV 4:2:0
"thermal.mp4", # Output file in MP4 container
]
subprocess.run(command)
print("to play: ffplay thermal.mp4")