renamed

client/clients.py

@@ -0,0 +1,130 @@
+#!/usr/bin/env python
+
+import asyncio
+from dataclasses import dataclass
+import os
+from socket import timeout
+from PIL import Image
+import json
+from multiprocessing import Pool
+import random
+import time
+from matplotlib import pyplot as plt
+import numpy as np
+
+import websockets
+
+@dataclass
+class pixel:
+    x: int
+    y: int
+    color: int
+    timestamp: int
+    userid: int
+
+
+def hex_to_rgb(h):
+    return tuple(int(h[i : i + 2], 16) for i in (0, 2, 4))
+
+
+hex_colors = [
+    "#FFFFFF",
+    "#E4E4E4",
+    "#888888",
+    "#222222",
+    "#FFA7D1",
+    "#E50000",
+    "#E59500",
+    "#A06A42",
+    "#E5D900",
+    "#94E044",
+    "#02BE01",
+    "#00D3DD",
+    "#0083C7",
+    "#0000EA",
+    "#CF6EE4",
+    "#820080",
+]
+rgb_colors = [hex_to_rgb(h[1:]) for h in hex_colors]
+
+
+def eucleadian_distance(rgb1, rgb2):
+    if len(rgb1) != len(rgb2):
+        raise ValueError
+    sum_part = np.sum([(i-j) ** 2 for i, j in zip(rgb1, rgb2)])
+    # return np.sqrt(sum_part) # technically correct, but we only care about rank not exact distance and sqrt is expensive
+    return sum_part
+
+
+def closest_match(rgb, color_map):
+    return min(
+        range(len(rgb_colors)), key=lambda i: eucleadian_distance(rgb, color_map[i])
+    )
+
+
+async def sender(target, img):
+    start_x = random.randint(0, 900)
+    start_y = random.randint(0, 900)
+    max_w, max_h, _ = img.shape
+    
+    async for websocket in websockets.connect(target + "/set"):
+        try:
+            for _ in range(int(max_h*max_w*1.3)):
+                rx = random.randint(0, max_w - 1)
+                ry = random.randint(0, max_h - 1)
+                if rx + start_x >= 1000 or ry + start_y >= 1000:
+                    continue
+                message = pixel(
+                    x=rx + start_x,
+                    y=ry + start_y,
+                    color=closest_match(img[rx][ry], rgb_colors),
+                    timestamp=int(time.time()),
+                    userid=1,
+                )
+                await websocket.send(json.dumps(message.__dict__))
+                succ = await websocket.recv()
+                if succ != "0":
+                    print(message, "was not set")
+            return
+        except websockets.ConnectionClosed:
+            print("reconnecting")
+
+
+async def client(target):
+    image = np.zeros(shape=[1000, 1000, 3], dtype=np.uint8)
+    async for websocket in websockets.connect(target + "/get"):
+        try:
+            x = pixel(**json.loads(await websocket.recv()))
+            image[x.x][x.y] = rgb_colors[x.color]
+            await websocket.send("1")
+        except websockets.ConnectionClosed:
+            continue
+
+
+def rescale(max_dimension, img):
+    w, h = img.size
+    maxi = max([w, h])
+    scale = max_dimension / maxi
+    return img.resize((int(scale * w), int(scale * h)), Image.ANTIALIAS)
+
+
+async def main(target):
+    images_folder_path = "./images"
+    while True:
+        image_paths = os.listdir(images_folder_path)
+        images = [
+            np.array(rescale(random.randint(100, 400), Image.open(f"{images_folder_path}/{image_path}")))
+            for image_path in image_paths
+        ]
+        coros = [sender(target, images[i % len(images)]) for i in range(len(images))]
+        _ = await asyncio.gather(*coros)
+
+
+def asyncMain(x, target):
+    asyncio.get_event_loop().run_until_complete(main(target))
+
+
+if __name__ == "__main__":
+    # with Pool(12) as p:
+    #    print(p.map(asyncMain, [() for _ in range(12)]))
+    asyncMain(0, target="ws://localhost:8080")

client/requirements.txt

@@ -0,0 +1,4 @@
+websockets
+numpy
+matplotlib
+opencv-python

server/server.go

@@ -0,0 +1,212 @@
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"fmt"
+	go_image "image"
+	"image/color"
+	"image/png"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"os"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+var upgrader = websocket.Upgrader{
+	ReadBufferSize:  4096,
+	WriteBufferSize: 4096,
+	CheckOrigin: func(r *http.Request) bool {
+		//origin := r.Header.Get("Origin")
+		return true
+	},
+}
+
+const (
+	// Time allowed to write a message to the peer.
+	writeWait = 10 * time.Second
+
+	// Time allowed to read the next pong message from the peer.
+	pongWait = 60 * time.Second
+
+	// Send pings to peer with this period. Must be less than pongWait.
+	pingPeriod = (pongWait * 9) / 10
+
+	// Maximum message size allowed from peer.
+	maxMessageSize = 512
+)
+
+var img = GetImage(1000, 1000)
+var tmpImage = GetImage(img.Width, img.Height)
+var diff = GetImage(img.Width, img.Height)
+
+func calcDiff() {
+	ticker := time.NewTicker(1 * time.Second)
+	for range ticker.C {
+		diff = tmpImage.GetDiff(&img)
+		copy(tmpImage.Pixels, img.Pixels)
+	}
+}
+
+func get(w http.ResponseWriter, r *http.Request) {
+	c, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		log.Print("error while upgrading", err)
+		return
+	}
+	fmt.Println("incoming connection")
+	c.SetReadLimit(maxMessageSize)
+	c.SetPongHandler(func(string) error { c.SetReadDeadline(time.Now().Add(pongWait)); return nil })
+
+	defer c.Close()
+	ticker := time.NewTicker(200 * time.Millisecond)
+	for range ticker.C {
+
+		for i := 0; i < int(diff.Width*diff.Height); i++ {
+
+			pix := diff.Pixels[i]
+
+			if pix.Pixel.UserID != 0 {
+				x := i / int(diff.Width)
+				y := i % int(diff.Height)
+				msg := Message{X: uint32(x), Y: uint32(y), Timestamp: pix.Pixel.Timestamp, UserID: pix.Pixel.UserID, Color: pix.Pixel.Color}
+				marshalMsg, err := json.Marshal(msg)
+				if err != nil {
+					log.Println("error while marshalling image", err)
+					break
+				}
+				err = c.WriteMessage(1, marshalMsg)
+				if err != nil {
+					log.Println("error while writing image", err)
+					break
+				}
+			}
+		}
+		if err := c.WriteMessage(websocket.PingMessage, []byte{}); err != nil {
+			return
+		}
+
+		copy(tmpImage.Pixels, img.Pixels)
+	}
+}
+func enableCors(w *http.ResponseWriter) {
+	(*w).Header().Set("Access-Control-Allow-Origin", "*")
+}
+func getAll(w http.ResponseWriter, r *http.Request) {
+	enableCors(&w)
+	w.Header().Set("Content-Type", "image/png")
+	colors := [16]color.Color{color.RGBA{255, 255, 255, 0xff}, color.RGBA{228, 228, 228, 0xff}, color.RGBA{136, 136, 136, 0xff}, color.RGBA{34, 34, 34, 0xff}, color.RGBA{255, 167, 209, 0xff}, color.RGBA{229, 0, 0, 0xff}, color.RGBA{229, 149, 0, 0xff}, color.RGBA{160, 106, 66, 0xff}, color.RGBA{229, 217, 0, 0xff}, color.RGBA{148, 224, 68, 0xff}, color.RGBA{2, 190, 1, 0xff}, color.RGBA{0, 211, 221, 0xff}, color.RGBA{0, 131, 199, 0xff}, color.RGBA{0, 0, 234, 0xff}, color.RGBA{207, 110, 228, 0xff}, color.RGBA{130, 0, 128, 0xff}}
+	upLeft := go_image.Point{0, 0}
+	lowRight := go_image.Point{int(img.Width), int(img.Height)}
+
+	png_img := go_image.NewRGBA(go_image.Rectangle{upLeft, lowRight})
+
+	for x := uint32(0); x < img.Width; x++ {
+		for y := uint32(0); y < img.Height; y++ {
+			png_img.Set(int(y), int(x), colors[img.Pixels[x*img.Width+y].Pixel.Color])
+		}
+	}
+	png.Encode(w, png_img)
+}
+
+func sendPing(ticker *time.Ticker, c *websocket.Conn, mutex *sync.Mutex) {
+	for range ticker.C {
+		mutex.Lock()
+		defer mutex.Unlock()
+		if err := c.WriteMessage(websocket.PingMessage, []byte{}); err != nil {
+			fmt.Println("Error while sending ping")
+			return
+		}
+	}
+}
+
+func set(w http.ResponseWriter, r *http.Request) {
+	c, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		log.Print("error while upgrading", err)
+		return
+	}
+	c.SetReadLimit(maxMessageSize)
+	c.SetPongHandler(func(string) error { c.SetReadDeadline(time.Now().Add(pongWait)); return nil })
+
+	ticker := time.NewTicker(pingPeriod / 2)
+	defer c.Close()
+	defer ticker.Stop()
+	mutex := sync.Mutex{}
+	go sendPing(ticker, c, &mutex)
+
+	for {
+
+		mt, msg, err := c.ReadMessage()
+		if err != nil {
+			log.Println("read:", err)
+			break
+		}
+		if mt == websocket.PingMessage {
+			continue
+		}
+		message := Message{}
+		json.Unmarshal(msg, &message)
+
+		status := img.SetPixel(message)
+		mutex.Lock()
+
+		err = c.WriteMessage(1, []byte(strconv.Itoa(status)))
+		mutex.Unlock()
+	}
+}
+
+func loadState(img *image, path string) {
+	stateJSON, err := os.Open(path)
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+	defer stateJSON.Close()
+	byteValue, _ := ioutil.ReadAll(stateJSON)
+
+	json.Unmarshal(byteValue, &img)
+}
+
+func saveState(img *image, path string, period time.Duration) {
+	ticker := time.NewTicker(period * time.Second)
+
+	for range ticker.C {
+		imgJSON, _ := json.Marshal(img)
+		file, err := os.Create(path)
+
+		if err != nil {
+			return
+		}
+		defer file.Close()
+		file.WriteString(string(imgJSON))
+		if err != nil {
+			fmt.Println("Could not save state")
+			fmt.Println(err)
+		}
+	}
+}
+
+func main() {
+	var addr = flag.String("addr", "localhost:8080", "http service address")
+
+	flag.Parse()
+	log.SetFlags(0)
+	log.Println("starting server on", *addr)
+
+	cachePath := "./state.json"
+
+	loadState(&img, cachePath)
+	go saveState(&img, cachePath, 10)
+	go calcDiff()
+	http.HandleFunc("/get", get)
+	http.HandleFunc("/getAll", getAll)
+	http.HandleFunc("/set", set)
+
+	log.Fatal(http.ListenAndServe(*addr, nil))
+}

server/util.go

@@ -0,0 +1,82 @@
+package main
+
+import (
+	"fmt"
+	"sync"
+)
+
+type Message struct {
+	X         uint32 `json:"x"`
+	Y         uint32 `json:"y"`
+	Color     uint8  `json:"color"`
+	Timestamp int64  `json:"timestamp"`
+	UserID    uint64 `json:"userid"`
+}
+
+type pixel struct {
+	Color     uint8  `json:"c"`
+	Timestamp int64  `json:"ts"`
+	UserID    uint64 `json:"uid"`
+}
+
+type pixelContainer struct {
+	Pixel pixel `json:"p"`
+	mutex sync.Mutex
+}
+
+type image struct {
+	Width  uint32           `json:"Width"`
+	Height uint32           `json:"Height"`
+	Pixels []pixelContainer `json:"Pixels"`
+	Mutex  sync.Mutex
+}
+
+func GetImage(w uint32, h uint32) image {
+	Pixels := make([]pixelContainer, w*h)
+	for i := 0; i < int(w*h); i++ {
+		Pixels[i] = pixelContainer{Pixel: pixel{Color: 0, Timestamp: 0, UserID: 0}, mutex: sync.Mutex{}}
+	}
+	return image{Width: w, Height: h, Pixels: Pixels, Mutex: sync.Mutex{}}
+}
+
+func (p *pixelContainer) setColor(color uint8, timestamp int64, userid uint64) {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+	if timestamp > p.Pixel.Timestamp {
+		p.Pixel.Color = color
+		p.Pixel.Timestamp = timestamp
+		p.Pixel.UserID = userid
+	}
+}
+
+func (img *image) SetPixel(message Message) int {
+	if message.X >= img.Width || message.Y >= img.Height || message.X < 0 || message.Y < 0 {
+		fmt.Printf("User %d tried accessing out of bounds \n", message.UserID)
+		return 1
+	}
+	if message.Color > 15 || message.Color < 0 {
+		fmt.Printf("User %d tried setting non existent color %d \n", message.UserID, message.Color)
+		return 1
+	}
+	pos := uint32(message.X)*uint32(img.Width) + uint32(message.Y)
+	img.Pixels[pos].setColor(message.Color, message.Timestamp, message.UserID)
+	return 0
+}
+
+func comparePixels(pixel1 *pixelContainer, pixel2 *pixelContainer) bool {
+	return pixel1.Pixel.Color == pixel2.Pixel.Color &&
+		pixel1.Pixel.Timestamp == pixel2.Pixel.Timestamp &&
+		pixel1.Pixel.UserID == pixel2.Pixel.UserID
+}
+
+func (img *image) GetDiff(img2 *image) image {
+	diff := GetImage(img.Width, img.Height)
+	for i := 0; i < int(img.Width*img.Height); i++ {
+		if !comparePixels(&img.Pixels[i], &img2.Pixels[i]) {
+			diff.Pixels[i].Pixel.Color = img2.Pixels[i].Pixel.Color
+			diff.Pixels[i].Pixel.UserID = img2.Pixels[i].Pixel.UserID
+			diff.Pixels[i].Pixel.Timestamp = img2.Pixels[i].Pixel.Timestamp
+		}
+	}
+	return diff
+}