cptlobster/video-streaming-poc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor image application into packet classes

Browse Source
This commit is contained in:
Camryn Thomas 2025-01-31 22:05:57 -06:00
parent 3e7ba62d3a
commit 98cdaad09e
Signed by: cptlobster
GPG Key ID: 33D607425C830B4C
2 changed files with 43 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
common.py
View File

@ -1,13 +1,28 @@
from abc import ABC, abstractmethod
import numpy as np
from uuid import UUID
class StdPacket:
size = 16 + 4 + 4 + 768
class Packet(ABC):
size: int
def __init__(self, uuid: UUID, x: int, y: int, array: np.ndarray):
self.uuid = uuid
self.x = x
self.y = y
self.array = array
@abstractmethod
def to_bytestr(self) -> bytes:
pass
@abstractmethod
def apply(self, image: np.ndarray) -> np.ndarray:
pass
class StdPacket(Packet):
size = 16 + 4 + 4 + 768
def __init__(self, uuid: UUID, x: int, y: int, array: np.ndarray):
super().__init__(uuid, x, y, array)
def to_bytestr(self) -> bytes:
bytestr = b""
@ -17,6 +32,13 @@ class StdPacket:
bytestr += self.array.tobytes()
return bytestr
def apply(self, image: np.ndarray) -> np.ndarray:
x = self.x
y = self.y
arr = self.array
image[y:y + 16, x:x + 16] = arr
return image
def from_bytes_std(b: bytes) -> StdPacket:
uuid = UUID(bytes = b[0:16])
x = int.from_bytes(b[16:20], signed = False)
@ -26,14 +48,11 @@ def from_bytes_std(b: bytes) -> StdPacket:
return StdPacket(uuid, x, y, array)
class InterlacedPacket:
class InterlacedPacket(Packet):
size = 16 + 4 + 4 + 4 + 768
def __init__(self, uuid: UUID, x: int, y: int, even: bool, array: np.ndarray):
self.uuid = uuid
self.x = x
self.y = y
super().__init__(uuid, x, y, array)
self.even = even
self.array = array
def to_bytestr(self) -> bytes:
bytestr = b""
@ -44,6 +63,13 @@ class InterlacedPacket:
bytestr += self.array.tobytes()
return bytestr
def apply(self, image: np.ndarray) -> np.ndarray:
x = self.x
y = self.y
arr = self.array
image[y + self.even:y + 32:2, x:x + 16] = arr
return image
def from_bytes_int(b: bytes) -> InterlacedPacket:
uuid = UUID(bytes=b[0:16])
@ -54,15 +80,12 @@ def from_bytes_int(b: bytes) -> InterlacedPacket:
return InterlacedPacket(uuid, x, y, even, array)
class DoublyInterlacedPacket:
class DoublyInterlacedPacket(Packet):
size = 16 + 4 + 4 + 4 + 768
def __init__(self, uuid: UUID, x: int, y: int, even_x: bool, even_y: bool, array: np.ndarray):
self.uuid = uuid
self.x = x
self.y = y
super().__init__(uuid, x, y, array)
self.even_x = even_x
self.even_y = even_y
self.array = array
def to_bytestr(self) -> bytes:
bytestr = b""
@ -74,6 +97,13 @@ class DoublyInterlacedPacket:
bytestr += self.array.tobytes()
return bytestr
def apply(self, image: np.ndarray) -> np.ndarray:
x = self.x
y = self.y
arr = self.array
image[y + self.even_y:y + 32:2, x + self.even_x:x + 32:2] = arr
return image
def from_bytes_dint(b: bytes) -> DoublyInterlacedPacket:
uuid = UUID(bytes=b[0:16])

5
server.py
View File

@ -21,10 +21,7 @@ class Client:
self.frame = np.ndarray((HEIGHT, WIDTH, 3), dtype=np.uint8)
def update(self, pkt: DoublyInterlacedPacket):
x = pkt.x
y = pkt.y
arr = pkt.array
self.frame[y + pkt.even_y:y + 32:2, x + pkt.even_x:x + 32:2] = arr
self.frame = pkt.apply(self.frame)
self.last_updated = datetime.now()