from abc import ABC, abstractmethod import numpy as np from uuid import UUID # The basic structure of the packet is as follows: # FFFF FFFF FFFF FFFF FFFF FFFF FFFF FFFF FFFF FFFF FFFF FFFF FFFF ... FFFF # | uuid | | | | # | | | | image data: contains the 16x16 image slice bitpacked from a NumPy array # | | | y: the y position of this packet in the original image # | | x: the x position of this packet in the original image # | uuid: matches the packet to the requesting client # Other packet types may change this structure. Need to standardize this somehow. class Packet(ABC): """Generic structure for a video streaming packet. Contains a slice of the full image, which will be reconstructed by the server.""" 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: """Convert a packet object into a bytestring.""" pass @abstractmethod def apply(self, image: np.ndarray) -> np.ndarray: """Apply this packet to an image.""" pass class StdPacket(Packet): """A standard packet with no interlacing. Sends a 16x16 chunk of an image.""" 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"" bytestr += self.uuid.bytes bytestr += self.x.to_bytes(length=4, signed = False) bytestr += self.y.to_bytes(length=4, signed = False) 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: """Convert a byte string obtained via UDP into a packet object.""" uuid = UUID(bytes = b[0:16]) x = int.from_bytes(b[16:20], signed = False) y = int.from_bytes(b[20:24], signed = False) array = np.frombuffer(b[24:], np.uint8).reshape(16, 16, 3) return StdPacket(uuid, x, y, array) class InterlacedPacket(Packet): """A packet with horizontal interlacing. Sends half of a 16x32 chunk of an image, every other row""" size = 16 + 4 + 4 + 4 + 768 def __init__(self, uuid: UUID, x: int, y: int, even: bool, array: np.ndarray): super().__init__(uuid, x, y, array) self.even = even def to_bytestr(self) -> bytes: bytestr = b"" bytestr += self.uuid.bytes bytestr += self.x.to_bytes(length=4, signed=False) bytestr += self.y.to_bytes(length=4, signed=False) bytestr += self.even.to_bytes(length=4) 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: """Convert a byte string obtained via UDP into a packet object.""" uuid = UUID(bytes=b[0:16]) x = int.from_bytes(b[16:20], signed=False) y = int.from_bytes(b[20:24], signed=False) even = bool.from_bytes(b[24:28]) array = np.frombuffer(b[28:], np.uint8).reshape(16, 16, 3) return InterlacedPacket(uuid, x, y, even, array) class DoublyInterlacedPacket(Packet): """A packet with horizontal interlacing. Sends one quarter of a 32x32 chunk of an image. This will alternate rows and columns based on the value of even_x and even_y.""" size = 16 + 4 + 4 + 4 + 768 def __init__(self, uuid: UUID, x: int, y: int, even_x: bool, even_y: bool, array: np.ndarray): super().__init__(uuid, x, y, array) self.even_x = even_x self.even_y = even_y def to_bytestr(self) -> bytes: bytestr = b"" bytestr += self.uuid.bytes bytestr += self.x.to_bytes(length=4, signed=False) bytestr += self.y.to_bytes(length=4, signed=False) bytestr += self.even_x.to_bytes(length=2) bytestr += self.even_y.to_bytes(length=2) 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: """Convert a byte string obtained via UDP into a packet object.""" uuid = UUID(bytes=b[0:16]) x = int.from_bytes(b[16:20], signed=False) y = int.from_bytes(b[20:24], signed=False) even_x = bool.from_bytes(b[24:26]) even_y = bool.from_bytes(b[26:28]) array = np.frombuffer(b[28:], np.uint8).reshape(16, 16, 3) return DoublyInterlacedPacket(uuid, x, y, even_x, even_y, array) class TiledImagePacket(Packet): """Distributed selection from image.""" size = 16 + 4 + 4 + 768 def __init__(self, uuid: UUID, x: int, y: int, width: int, height: int, array: np.ndarray): super().__init__(uuid, x, y, array) self.width = width self.height = height self.xslice = width // 16 self.yslice = height // 16 def to_bytestr(self) -> bytes: bytestr = b"" bytestr += self.uuid.bytes bytestr += self.x.to_bytes(length=4, signed=False) bytestr += self.y.to_bytes(length=4, signed=False) 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.height:self.yslice, x:self.width:self.xslice] = arr return image def from_bytes_tiled(b: bytes) -> TiledImagePacket: """Convert a byte string obtained via UDP into a packet object.""" uuid = UUID(bytes = b[0:16]) x = int.from_bytes(b[16:20], signed = False) y = int.from_bytes(b[20:24], signed = False) array = np.frombuffer(b[24:], np.uint8).reshape(16, 16, 3) return TiledImagePacket(uuid, x, y, 640, 480, array)