noe kreativt

WeatherGraph-master/README.md

@@ -0,0 +1,24 @@
+# WeatherGrapher
+
+A system for generating temperature graphs.
+
+## Components
+
+`tempgraph`
+
+The full executable which downloads data, parses it, displays it and generates a graph.
+This component orchestrates communication between the other components.
+
+`strip_spaces.sh`
+
+Removes extra spaces between elements.
+
+`get_temperature.sh`
+
+Get only the first two columns.
+
+`display_.py`
+
+Generate a view of the data as it is processed, and save the final data to file.
+
+``

WeatherGraph-master/display_results.py

@@ -0,0 +1,118 @@
+#!/usr/bin/env python3
+'''
+Reads a two-column space-seperated input on STDIN
+With the first column as HHMMSS, and the 2nd as
+a value
+Plots the values as a function of time
+Saves the data to temperatures.bmp
+'''
+
+import sys
+import socket
+
+import pygame
+
+class DisplayResult:
+    '''
+    Class for plotting data using pygame, scaled to fit to the
+    data
+    '''
+    def __init__(self):
+        pygame.init()
+        self._screen_size = 800
+        self._screen = pygame.display.set_mode((self._screen_size,
+                                                self._screen_size))
+        self._point = pygame.Surface((3,3))
+        self._point.fill((255,0,0))
+        self.points = []
+        self.minmax_x = [999999, -999999]
+        self.minmax_y = [999999, -999999]
+
+    def _update_minmax_points(self, new_point):
+        x, y = new_point
+        self.minmax_x[0] = min(x, self.minmax_x[0])
+        self.minmax_x[1] = max(x, self.minmax_x[1])
+        self.minmax_y[0] = min(y, self.minmax_y[0])
+        self.minmax_y[1] = max(y, self.minmax_y[1])
+
+    def _get_seconds(self, time):
+        hour = int(time[:2])
+        minute = int(time[2:4])
+        second = int(time[4:])
+        return second + (minute * 60) + (hour * 60 * 60)
+
+    def _get_datapoint(self, line):
+        timestamp, y = line.split()
+        y = float(y)
+        x = self._get_seconds(timestamp)
+        return (x, y)
+
+    def _update_scaling_factors(self):
+        try:
+            self.x_factor = self._screen_size / (self.minmax_x[1] - self.minmax_x[0])
+        except ZeroDivisionError:
+            self.x_factor = 1
+        self.x_constant = self.minmax_x[0]
+
+        try:
+            self.y_factor = self._screen_size / (self.minmax_y[1] - self.minmax_y[0])
+        except ZeroDivisionError:
+            self.y_factor = 1
+        self.y_constant = self.minmax_y[0]
+
+    def _draw_screen(self):
+        self._screen.fill((0,0,0))
+        for p in self.points:
+            real_x = int((p[0] - self.x_constant) * self.x_factor)
+            real_y = int((p[1] - self.y_constant) * self.y_factor)
+            self._screen.blit(self._point, (real_x, real_y))
+        pygame.display.flip()
+
+    def save_image(self, filename):
+        '''
+        Save the current graph to file.
+        '''
+        pygame.image.save(self._screen, filename)
+
+    def new_data(self, line):
+        '''
+        Add a new data point to the graph.
+        '''
+        self.points.append(self._get_datapoint(line))
+        self._update_minmax_points(self.points[-1])
+        self._update_scaling_factors()
+        self._draw_screen()
+
+    def get_data(self):
+        s = socket.create_connection(('weather.cs.uit.no', 44102))
+        done = False
+        full_data = ""
+        #print("Socket created at {}".format(s))
+        while not done:
+            data = s.recv()
+            #print("Data: ", data)
+            if data == "":
+                break
+            full_data = full_data + data
+        socket.close(s)
+        return full_data
+
+    def strip_spaces(self, input_string):
+        input_string = " ".join(input_string.split())
+        return input_string
+
+
+def mainloop():
+    A = DisplayResult()
+    print("A is ok")
+    for line in sys.stdin:
+        A.new_data(line)
+        #A.strip_spaces(A.new_data)
+        print(A.new_data)
+    A.save_image("temperatures.bmp")
+    print("about to exit!")
+    sys.exit()
+
+if __name__ == "__main__":
+    mainloop()
+

WeatherGraph-master/get_temperatures.sh

@@ -0,0 +1,10 @@
+#!/usr/bin/env bash
+
+# Gets the first two columns of a space-seperated input on STDIN
+# Outputs the first two columns on STDOUT
+
+while read line
+do
+	a="$(echo "$line" | cut -d ' ' -f 1,2)"
+	echo "$a"
+done < "${1:-/dev/stdin}"

WeatherGraph-master/strip_spaces.sh

@@ -0,0 +1,10 @@
+#!/usr/bin/env bash
+
+# Reads data from STDIN, and strips repeated spaces
+# Outputs the stdin data with spaces removed to STDOUT
+
+while read line
+do
+	a="$(echo "$line" | sed -r 's/ +/ /g')"
+	echo "$a"
+done < "${1:-/dev/stdin}"

WeatherGraph-master/tempgraph

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+
+nc -Nd weather.cs.uit.no 44102 | ./strip_spaces.sh | ./get_temperatures.sh |./display_results.py

WeatherGraph-master/test_display.py

@@ -0,0 +1,61 @@
+import unittest
+import unittest.mock as m
+import subprocess
+
+import display_results
+
+class FakeConnection:
+    def __init__(self):
+        self.lines = ["000031    1.6    2.2  254.2   -3.3   72.2  "\
+                      "993.3  -15.8    0.0    0.0    0.0    0.0    " \
+                      "0.0\n000131    1.4    1.9  253.5   -3.2   72.9  "\
+                      "993.5  -21.0    0.0    0.0    0.0    0.0    0.0",
+                      ""]
+        self.recv_called = 0
+
+    def recv(self, *args):
+        self.recv_called += 1
+        return self.lines.pop(0)
+
+    def __next__(self):
+        return self
+
+
+class TestDisplay(unittest.TestCase):
+    def setUp(self):
+        self.disp = display_results.DisplayResult()
+        self.disp.draw_screen = lambda x: None
+
+    def test_update_points(self):
+        self.disp.new_data("000010 123")
+        self.assertEqual(self.disp.points[0][0], 10)
+        self.assertEqual(self.disp.points[0][1], 123)
+        self.disp.new_data("000012 12")
+        self.assertEqual(self.disp.points[-1][0], 12)
+        self.assertEqual(self.disp.points[-1][1], 12)
+        self.assertEqual(self.disp.minmax_x[0], 10)
+        self.assertEqual(self.disp.minmax_x[1], 12)
+        self.assertEqual(self.disp.minmax_y[0], 12)
+        self.assertEqual(self.disp.minmax_y[1], 123)
+
+    def test_write_image(self):
+        self.disp.save_image("test.bmp")
+        val = subprocess.run(["md5sum", "test.bmp"], stdout=subprocess.PIPE).stdout
+        self.assertEqual(val, b'777e538381377589c8a6f9d6d53a2006  test.bmp\n')
+
+    @m.patch("socket.create_connection")
+    def test_get_data(self, socket):
+        fc = FakeConnection()
+        expected_result = "".join(fc.lines)
+        socket.side_effect = fc
+        data = self.disp.get_data()
+        self.assertEqual(fc.recv_called, 2)
+        self.assertEqual(expected_result, data)
+
+    def test_strip_spaces(self):
+        input_string = "hello   world    two"
+        output_string = "hello world two"
+        self.assertEqual(self.disp.strip_spaces(input_string), output_string)
+
+if __name__ == "__main__":
+    unittest.main()