Added Writer to openair (Turbo87#279)

aerofiles/openair/writer.py

@@ -0,0 +1,130 @@
+import math
+
+
+class Writer:
+
+    """
+    A higher-level writer for the OpenAir airspace file format::
+
+        with open('airspace.txt', 'wb') as fp:
+            writer = Writer(fp)
+
+    :param fp: file pointer to write to
+    :param encoding: the encoding used for the output
+
+    see `OpenAir file format specification
+    <http://www.winpilot.com/UsersGuide/UserAirspace.asp>`_
+
+    This class should be used to write records as described under Reader
+    into a file.
+
+        writer.write_record(record)
+
+    Currently on airspace records are implemented. Terrain is missing.
+
+    """
+
+    def reset_V(self):
+        self.V_X = None
+        self.V_D = 1
+
+    def __init__(self, fp=None, encoding='utf-8'):
+        self.fp = fp
+        self.encoding = encoding
+        self.reset_V()
+
+    def format_dms(self, decimal_degrees):
+        decimals, number = math.modf(decimal_degrees)
+        deg = int(number)
+        mnt = int(decimals * 60)
+        sec = round((decimal_degrees - deg - mnt / 60) * 3600.00)
+        if sec == 60:
+            sec = 0
+            mnt += 1
+            if mnt == 60:
+                mnt = 0
+                deg += 1
+        return deg, mnt, sec
+
+    def format_degree(self, v, width):
+        (degrees, minutes, seconds) = self.format_dms(abs(v))
+        result = "%0*d:%02d:%02d" % (width, degrees, minutes, seconds)
+        return result
+
+    def format_coord(self, P):
+        result = self.format_degree(abs(P[0]), 2) + " "
+        if P[0] >= 0:
+            result = result + "N "
+        else:
+            result = result + "S "
+
+        result = result + self.format_degree(abs(P[1]), 3) + " "
+        if P[1] >= 0:
+            result = result + "E"
+        else:
+            result = result + "W"
+        return result
+
+    def write_line(self, line):
+        self.fp.write((line + u'\r\n').encode(self.encoding))
+
+    def write_V_X(self, center):
+        if center != self.V_X:
+            self.V_X = center
+            center = self.format_coord(center)
+            self.write_line('V X=' + center)
+
+    def write_V_D(self, clockwise):
+        if clockwise != self.V_D:
+            self.V_D = clockwise
+            if clockwise:
+                D = "+"
+            else:
+                D = "-"
+            self.write_line('V D=' + D)
+
+    def write_DC(self, element):
+        self.write_V_X(element["center"])
+        self.write_line('DC %s' % (str(element["radius"])))
+
+    def write_DA(self, element):
+        self.write_V_X(element["center"])
+        self.write_V_D(element["clockwise"])
+        self.write_line('DA %s,%s,%s' % (str(element["radius"]), str(element["start"]), str(element["end"])))
+
+    def write_DB(self, element):
+        self.write_V_X(element["center"])
+        self.write_V_D(element["clockwise"])
+        start = self.format_coord(element["start"])
+        end = self.format_coord(element["end"])
+        self.write_line('DB %s, %s' % (start, end))
+
+    def write_DP(self, element):
+        location = self.format_coord(element["location"])
+        self.write_line('DP %s' % (location))
+
+    def write_airspace_element(self, element):
+        if element["type"] == "point":
+            self.write_DP(element)
+        elif element["type"] == "circle":
+            self.write_DC(element)
+        elif element["type"] == "arc":
+            if "radius" in element:
+                self.write_DA(element)
+            else:
+                self.write_DB(element)
+
+    def write_airspace(self, record):
+        self.reset_V()
+        self.write_line('AC ' + record["class"])
+        self.write_line('AN ' + record["name"])
+        self.write_line('AH ' + record["ceiling"])
+        self.write_line('AL ' + record["floor"])
+        for element in record["elements"]:
+            self.write_airspace_element(element)
+
+    def write_record(self, record):
+        if record["type"] == "airspace":
+            self.write_airspace(record)
+        else:
+            raise ValueError('unknown record type: ' + record["type"])

tests/openair/test_writer.py

@@ -0,0 +1,118 @@
+from io import BytesIO
+from os import path
+
+from aerofiles.openair.writer import Writer
+
+import pytest
+
+DATA = path.join(path.dirname(path.realpath(__file__)), 'data')
+
+
+# Fixtures ####################################################################
+
+@pytest.fixture()
+def output():
+    return BytesIO()
+
+
+@pytest.fixture()
+def writer(output):
+    return Writer(output)
+
+# Tests #######################################################################
+
+
+def test_write_line(writer):
+    writer.write_line('line')
+    assert writer.fp.getvalue() == b'line\r\n'
+
+
+def test_write_DP(writer):
+    element = {"location": [-39.58333, -118.98888]}
+    writer.write_DP(element)
+    assert writer.fp.getvalue() == b'DP 39:35:00 S 118:59:20 W\r\n'
+
+
+def test_write_DC(writer):
+    element = {"center": [39.58333, 118.98888], "radius": 10}
+    writer.write_DC(element)
+    assert writer.fp.getvalue() == b'V X=39:35:00 N 118:59:20 E\r\nDC 10\r\n'
+
+
+def test_write_DA(writer):
+    element = {"center": [39.58333, 118.98888],
+               "radius": 10.1,
+               "start": 44.9, "end": 88, "clockwise": True}
+    writer.write_DA(element)
+
+    # Make sure, that "V X=" is not repeated but "V D=" is used:
+    element["clockwise"] = False
+    writer.write_DA(element)
+
+    assert writer.fp.getvalue() == b'V X=39:35:00 N 118:59:20 E\r\nDA 10.1,44.9,88\r\nV D=-\r\nDA 10.1,44.9,88\r\n'
+
+
+def test_write_DB(writer):
+    element = {"center": [39.495, -119.775],
+               "start": [39.61333, -119.76833],
+               "end": [39.49833, -119.60166],
+               "clockwise": True}
+    writer.write_DB(element)
+
+    assert writer.fp.getvalue() == b'V X=39:29:42 N 119:46:30 W\r\nDB 39:36:48 N 119:46:06 W, 39:29:54 N 119:36:06 W\r\n'
+
+
+def test_write_record(writer):
+    record = {
+        "type": "airspace",
+        "class": "C",
+        "name": "RENO",
+        "floor": "7200 ft",
+        "ceiling": "8400 ft",
+        "elements": [{
+            "type": "arc",
+            "center": [39.495, -119.775],
+            "radius": 10,
+            "start": 270,
+            "end": 290,
+            "clockwise": True
+        }, {
+            "type": "arc",
+            "center": [39.495, -119.775],
+            "radius": 7,
+            "start": 290,
+            "end": 320,
+            "clockwise": False
+        }, {
+            "type": "arc",
+            "center": [39.495, -119.775],
+            "start": [39.61333, -119.76833],
+            "end": [39.49833, -119.60166],
+            "clockwise": True
+        }, {
+            "type": "point",
+            "location": [39.495, -119.775]
+        }, {
+            "type": "circle",
+            "center": [39.495, -119.775],
+            "radius": 5,
+        }]
+    }
+    writer.write_record(record)
+    assert writer.fp.getvalue() == b'\r\n'.join([
+        b'AC C',
+        b'AN RENO',
+        b'AH 8400 ft',
+        b'AL 7200 ft',
+        b'V X=39:29:42 N 119:46:30 W',
+        b'DA 10,270,290',
+        b'V D=-',
+        b'DA 7,290,320',
+        b'V D=+',
+        b'DB 39:36:48 N 119:46:06 W, 39:29:54 N 119:36:06 W',
+        b'DP 39:29:42 N 119:46:30 W',
+        b'DC 5',
+    ]) + b'\r\n'
+
+
+# Assertions ##################################################################