feat: add exif based image rotation

spyglass/exif.py

@@ -0,0 +1,47 @@
+def create_exif_header(orientation: int):
+    if orientation <= 0:
+        return None
+
+    return b''.join([
+        b'\xFF\xD8',  # Start of Image (SOI) marker
+        b'\xFF\xE1',  # APP1 marker
+        b'\x00\x62',  # Length of APP 1 segment (98 bytes)
+        b'\x45\x78\x69\x66',  # EXIF identifier ("Exif" in ASCII)
+        b'\x00\x00',  # Padding bytes
+        # TIFF header (with big-endian indicator)
+        b'\x4D\x4D',  # Big endian
+        b'\x00\x2A',  # TIFF magic number
+        b'\x00\x00\x00\x08',  # Offset to first IFD (8 bytes)
+        # Image File Directory (IFD)
+        b'\x00\x05',  # Number of entries in the IFD (5)
+        # v-- Orientation tag (tag number = 0x0112, type = USHORT, count = 1)
+        b'\x01\x12', b'\x00\x03', b'\x00\x00\x00\x01',
+        b'\x00', orientation.to_bytes(1, 'big'), b'\x00\x00',  # Tag data
+        # v-- XResolution tag (tag number = 0x011A, type = UNSIGNED RATIONAL, count = 1)
+        b'\x01\x1A', b'\x00\x05', b'\x00\x00\x00\x01',
+        b'\x00\x00\x00\x4A',  # Tag data (address)
+        # v-- YResolution tag (tag number = 0x011B, type = UNSIGNED RATIONAL, count = 1)
+        b'\x01\x1B', b'\x00\x05', b'\x00\x00\x00\x01',
+        b'\x00\x00\x00\x52',  # Tag data (address)
+        # v-- ResolutionUnit tag (tag number = 0x0128, type = USHORT, count = 1)
+        b'\x01\x28', b'\x00\x03', b'\x00\x00\x00\x01',
+        b'\x00\x02\x00\x00',  # 2 - Inch
+        # v-- YCbCrPositioning tag (tag number = 0x0213, type = USHORT, count = 1)
+        b'\x02\x13', b'\x00\x03', b'\x00\x00\x00\x01',
+        b'\x00\x01\x00\x00',  # center of pixel array
+        b'\x00\x00\x00\x00',  # Offset to next IFD 0
+        b'\x00\x00\x00\x48\x00\x00\x00\x01',  # XResolution value
+        b'\x00\x00\x00\x48\x00\x00\x00\x01'  # YResolution value
+    ])
+
+
+option_to_exif_orientation_map = {
+    'h': 1,
+    'mh': 2,
+    'r180': 3,
+    'mv': 4,
+    'mhr270': 5,
+    'r90': 6,
+    'mhr90': 7,
+    'r270': 8
+}

spyglass/server.py

@@ -3,6 +3,8 @@
 import socketserver
 from http import server
 from threading import Condition
+
+from spyglass.exif import create_exif_header
 from . import logger
 
 
@@ -22,8 +24,10 @@ class StreamingServer(socketserver.ThreadingMixIn, server.HTTPServer):
     daemon_threads = True
 
 
-def run_server(bind_address, port, output, stream_url='/stream', snapshot_url='/snapshot', orientation_exif = 0):
-    class StreamingHandler(server.BaseHTTPRequestHandler):        
+def run_server(bind_address, port, output, stream_url='/stream', snapshot_url='/snapshot', orientation_exif=0):
+    exif_header = create_exif_header(orientation_exif)
+
+    class StreamingHandler(server.BaseHTTPRequestHandler):
         def do_GET(self):
             if self.path == stream_url:
                 self.start_streaming()
@@ -44,14 +48,13 @@ def start_streaming(self):
                         output.condition.wait()
                         frame = output.frame
                     self.wfile.write(b'--FRAME\r\n')
-                    if orientation_exif <= 0:
+                    if exif_header is None:
                         self.send_jpeg_content_headers(frame)
                         self.end_headers()
                         self.wfile.write(frame)
                         self.wfile.write(b'\r\n')
                     else:
-                        exif_header = self.get_orientation_exif_header(orientation_exif)
-                        self.send_jpeg_content_headers(frame, len(exif_header)-2)
+                        self.send_jpeg_content_headers(frame, len(exif_header) - 2)
                         self.end_headers()
                         self.wfile.write(exif_header)
                         self.wfile.write(frame[2:])
@@ -71,8 +74,7 @@ def send_snapshot(self):
                     self.end_headers()
                     self.wfile.write(frame)
                 else:
-                    exif_header = self.get_orientation_exif_header(orientation_exif)
-                    self.send_jpeg_content_headers(frame, len(exif_header)-2)
+                    self.send_jpeg_content_headers(frame, len(exif_header) - 2)
                     self.end_headers()
                     self.wfile.write(exif_header)
                     self.wfile.write(frame[2:])
@@ -88,18 +90,7 @@ def send_default_headers(self):
 
         def send_jpeg_content_headers(self, frame, extra_len=0):
             self.send_header('Content-Type', 'image/jpeg')
-            self.send_header('Content-Length', len(frame)+extra_len)
-
-        def get_orientation_exif_header(self, orientation_exif):
-            return b''.join([
-                    b'\xFF\xD8\xFF\xE1\x00\x62\x45\x78\x69\x66\x00\x00\x4D\x4D\x00\x2A',
-                    b'\x00\x00\x00\x08\x00\x05\x01\x12\x00\x03\x00\x00\x00\x01\x00',
-                    (orientation_exif).to_bytes(1, 'big'),
-                    b'\x00\x00\x01\x1A\x00\x05\x00\x00\x00\x01\x00\x00\x00\x4A\x01\x1B',
-                    b'\x00\x05\x00\x00\x00\x01\x00\x00\x00\x52\x01\x28\x00\x03\x00\x00',
-                    b'\x00\x01\x00\x02\x00\x00\x02\x13\x00\x03\x00\x00\x00\x01\x00\x01',
-                    b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x48\x00\x00\x00\x01\x00\x00',
-                    b'\x00\x48\x00\x00\x00\x01'])
+            self.send_header('Content-Length', str(len(frame) + extra_len))
 
     logger.info('Server listening on %s:%d', bind_address, port)
     logger.info('Streaming endpoint: %s', stream_url)

tests/test_exif.py

@@ -0,0 +1,17 @@
+import pytest
+
+
+@pytest.mark.parametrize("input_value, expected_output", [
+    ('h', 1),
+    ('mh', 2),
+    ('r180', 3),
+    ('mv', 4),
+    ('mhr270', 5),
+    ('r90', 6),
+    ('mhr90', 7),
+    ('r270', 8),
+])
+def test_option_to_exif_orientation_map(input_value, expected_output):
+    from spyglass.exif import option_to_exif_orientation_map
+    orientation_value = option_to_exif_orientation_map[input_value]
+    assert orientation_value == expected_output