from __future__ import unicode_literals

import re
from decimal import Decimal

from django.contrib.gis.db.models import functions
from django.contrib.gis.geos import LineString, Point, Polygon, fromstr
from django.contrib.gis.measure import Area
from django.db import connection
from django.db.models import Sum
from django.test import TestCase, skipUnlessDBFeature
from django.utils import six

from ..utils import mysql, oracle, postgis, spatialite
from .models import City, Country, CountryWebMercator, State, Track


@skipUnlessDBFeature("gis_enabled")
class GISFunctionsTests(TestCase):
    """
    Testing functions from django/contrib/gis/db/models/functions.py.
    Several tests are taken and adapted from GeoQuerySetTest.
    Area/Distance/Length/Perimeter are tested in distapp/tests.

    Please keep the tests in function's alphabetic order.
    """
    fixtures = ['initial']

    def test_asgeojson(self):
        # Only PostGIS and SpatiaLite support GeoJSON.
        if not connection.ops.geojson:
            with self.assertRaises(NotImplementedError):
                list(Country.objects.annotate(json=functions.AsGeoJSON('mpoly')))
            return

        pueblo_json = '{"type":"Point","coordinates":[-104.609252,38.255001]}'
        houston_json = (
            '{"type":"Point","crs":{"type":"name","properties":'
            '{"name":"EPSG:4326"}},"coordinates":[-95.363151,29.763374]}'
        )
        victoria_json = (
            '{"type":"Point","bbox":[-123.30519600,48.46261100,-123.30519600,48.46261100],'
            '"coordinates":[-123.305196,48.462611]}'
        )
        chicago_json = (
            '{"type":"Point","crs":{"type":"name","properties":{"name":"EPSG:4326"}},'
            '"bbox":[-87.65018,41.85039,-87.65018,41.85039],"coordinates":[-87.65018,41.85039]}'
        )
        if spatialite:
            victoria_json = (
                '{"type":"Point","bbox":[-123.305196,48.462611,-123.305196,48.462611],'
                '"coordinates":[-123.305196,48.462611]}'
            )

        # Precision argument should only be an integer
        with self.assertRaises(TypeError):
            City.objects.annotate(geojson=functions.AsGeoJSON('point', precision='foo'))

        # Reference queries and values.
        # SELECT ST_AsGeoJson("geoapp_city"."point", 8, 0)
        # FROM "geoapp_city" WHERE "geoapp_city"."name" = 'Pueblo';
        self.assertEqual(
            pueblo_json,
            City.objects.annotate(geojson=functions.AsGeoJSON('point')).get(name='Pueblo').geojson
        )

        # SELECT ST_AsGeoJson("geoapp_city"."point", 8, 2) FROM "geoapp_city"
        # WHERE "geoapp_city"."name" = 'Houston';
        # This time we want to include the CRS by using the `crs` keyword.
        self.assertEqual(
            houston_json,
            City.objects.annotate(json=functions.AsGeoJSON('point', crs=True)).get(name='Houston').json
        )

        # SELECT ST_AsGeoJson("geoapp_city"."point", 8, 1) FROM "geoapp_city"
        # WHERE "geoapp_city"."name" = 'Houston';
        # This time we include the bounding box by using the `bbox` keyword.
        self.assertEqual(
            victoria_json,
            City.objects.annotate(
                geojson=functions.AsGeoJSON('point', bbox=True)
            ).get(name='Victoria').geojson
        )

        # SELECT ST_AsGeoJson("geoapp_city"."point", 5, 3) FROM "geoapp_city"
        # WHERE "geoapp_city"."name" = 'Chicago';
        # Finally, we set every available keyword.
        self.assertEqual(
            chicago_json,
            City.objects.annotate(
                geojson=functions.AsGeoJSON('point', bbox=True, crs=True, precision=5)
            ).get(name='Chicago').geojson
        )

    @skipUnlessDBFeature("has_AsGML_function")
    def test_asgml(self):
        # Should throw a TypeError when trying to obtain GML from a
        # non-geometry field.
        qs = City.objects.all()
        with self.assertRaises(TypeError):
            qs.annotate(gml=functions.AsGML('name'))
        ptown = City.objects.annotate(gml=functions.AsGML('point', precision=9)).get(name='Pueblo')

        if oracle:
            # No precision parameter for Oracle :-/
            gml_regex = re.compile(
                r'^<gml:Point srsName="SDO:4326" xmlns:gml="http://www.opengis.net/gml">'
                r'<gml:coordinates decimal="\." cs="," ts=" ">-104.60925\d+,38.25500\d+ '
                r'</gml:coordinates></gml:Point>'
            )
        else:
            gml_regex = re.compile(
                r'^<gml:Point srsName="EPSG:4326"><gml:coordinates>'
                r'-104\.60925\d+,38\.255001</gml:coordinates></gml:Point>'
            )

        self.assertTrue(gml_regex.match(ptown.gml))

        if postgis:
            self.assertIn(
                '<gml:pos srsDimension="2">',
                City.objects.annotate(gml=functions.AsGML('point', version=3)).get(name='Pueblo').gml
            )

    @skipUnlessDBFeature("has_AsKML_function")
    def test_askml(self):
        # Should throw a TypeError when trying to obtain KML from a
        # non-geometry field.
        with self.assertRaises(TypeError):
            City.objects.annotate(kml=functions.AsKML('name'))

        # Ensuring the KML is as expected.
        ptown = City.objects.annotate(kml=functions.AsKML('point', precision=9)).get(name='Pueblo')
        self.assertEqual('<Point><coordinates>-104.609252,38.255001</coordinates></Point>', ptown.kml)

    @skipUnlessDBFeature("has_AsSVG_function")
    def test_assvg(self):
        with self.assertRaises(TypeError):
            City.objects.annotate(svg=functions.AsSVG('point', precision='foo'))
        # SELECT AsSVG(geoapp_city.point, 0, 8) FROM geoapp_city WHERE name = 'Pueblo';
        svg1 = 'cx="-104.609252" cy="-38.255001"'
        # Even though relative, only one point so it's practically the same except for
        # the 'c' letter prefix on the x,y values.
        svg2 = svg1.replace('c', '')
        self.assertEqual(svg1, City.objects.annotate(svg=functions.AsSVG('point')).get(name='Pueblo').svg)
        self.assertEqual(svg2, City.objects.annotate(svg=functions.AsSVG('point', relative=5)).get(name='Pueblo').svg)

    @skipUnlessDBFeature("has_BoundingCircle_function")
    def test_bounding_circle(self):
        qs = Country.objects.annotate(circle=functions.BoundingCircle('mpoly')).order_by('name')
        self.assertAlmostEqual(qs[0].circle.area, 168.89, 2)
        self.assertAlmostEqual(qs[1].circle.area, 135.95, 2)

        qs = Country.objects.annotate(circle=functions.BoundingCircle('mpoly', num_seg=12)).order_by('name')
        self.assertAlmostEqual(qs[0].circle.area, 168.44, 2)
        self.assertAlmostEqual(qs[1].circle.area, 135.59, 2)

    @skipUnlessDBFeature("has_Centroid_function")
    def test_centroid(self):
        qs = State.objects.exclude(poly__isnull=True).annotate(centroid=functions.Centroid('poly'))
        tol = 1.8 if mysql else (0.1 if oracle else 0.00001)
        for state in qs:
            self.assertTrue(state.poly.centroid.equals_exact(state.centroid, tol))

        with self.assertRaisesMessage(TypeError, "'Centroid' takes exactly 1 argument (2 given)"):
            State.objects.annotate(centroid=functions.Centroid('poly', 'poly'))

    @skipUnlessDBFeature("has_Difference_function")
    def test_difference(self):
        geom = Point(5, 23, srid=4326)
        qs = Country.objects.annotate(diff=functions.Difference('mpoly', geom))
        # SpatiaLite and Oracle do something screwy with the Texas geometry.
        if spatialite or oracle:
            qs = qs.exclude(name='Texas')

        for c in qs:
            self.assertTrue(c.mpoly.difference(geom).equals(c.diff))

    @skipUnlessDBFeature("has_Difference_function", "has_Transform_function")
    def test_difference_mixed_srid(self):
        """Testing with mixed SRID (Country has default 4326)."""
        geom = Point(556597.4, 2632018.6, srid=3857)  # Spherical mercator
        qs = Country.objects.annotate(difference=functions.Difference('mpoly', geom))
        # SpatiaLite and Oracle do something screwy with the Texas geometry.
        if spatialite or oracle:
            qs = qs.exclude(name='Texas')
        for c in qs:
            self.assertTrue(c.mpoly.difference(geom).equals(c.difference))

    @skipUnlessDBFeature("has_Envelope_function")
    def test_envelope(self):
        countries = Country.objects.annotate(envelope=functions.Envelope('mpoly'))
        for country in countries:
            self.assertIsInstance(country.envelope, Polygon)

    @skipUnlessDBFeature("has_ForceRHR_function")
    def test_force_rhr(self):
        rings = (
            ((0, 0), (5, 0), (0, 5), (0, 0)),
            ((1, 1), (1, 3), (3, 1), (1, 1)),
        )
        rhr_rings = (
            ((0, 0), (0, 5), (5, 0), (0, 0)),
            ((1, 1), (3, 1), (1, 3), (1, 1)),
        )
        State.objects.create(name='Foo', poly=Polygon(*rings))
        st = State.objects.annotate(force_rhr=functions.ForceRHR('poly')).get(name='Foo')
        self.assertEqual(rhr_rings, st.force_rhr.coords)

    @skipUnlessDBFeature("has_GeoHash_function")
    def test_geohash(self):
        # Reference query:
        # SELECT ST_GeoHash(point) FROM geoapp_city WHERE name='Houston';
        # SELECT ST_GeoHash(point, 5) FROM geoapp_city WHERE name='Houston';
        ref_hash = '9vk1mfq8jx0c8e0386z6'
        h1 = City.objects.annotate(geohash=functions.GeoHash('point')).get(name='Houston')
        h2 = City.objects.annotate(geohash=functions.GeoHash('point', precision=5)).get(name='Houston')
        self.assertEqual(ref_hash, h1.geohash)
        self.assertEqual(ref_hash[:5], h2.geohash)

    @skipUnlessDBFeature("has_Intersection_function")
    def test_intersection(self):
        geom = Point(5, 23, srid=4326)
        qs = Country.objects.annotate(inter=functions.Intersection('mpoly', geom))
        for c in qs:
            if spatialite or (mysql and not connection.ops.uses_invalid_empty_geometry_collection) or oracle:
                # When the intersection is empty, some databases return None.
                expected = None
            else:
                expected = c.mpoly.intersection(geom)
            self.assertEqual(c.inter, expected)

    @skipUnlessDBFeature("has_IsValid_function")
    def test_isvalid(self):
        valid_geom = fromstr('POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))')
        invalid_geom = fromstr('POLYGON((0 0, 0 1, 1 1, 1 0, 1 1, 1 0, 0 0))')
        State.objects.create(name='valid', poly=valid_geom)
        State.objects.create(name='invalid', poly=invalid_geom)
        valid = State.objects.filter(name='valid').annotate(isvalid=functions.IsValid('poly')).first()
        invalid = State.objects.filter(name='invalid').annotate(isvalid=functions.IsValid('poly')).first()
        self.assertIs(valid.isvalid, True)
        self.assertIs(invalid.isvalid, False)

    @skipUnlessDBFeature("has_Area_function")
    def test_area_with_regular_aggregate(self):
        # Create projected country objects, for this test to work on all backends.
        for c in Country.objects.all():
            CountryWebMercator.objects.create(name=c.name, mpoly=c.mpoly)
        # Test in projected coordinate system
        qs = CountryWebMercator.objects.annotate(area_sum=Sum(functions.Area('mpoly')))
        # Some backends (e.g. Oracle) cannot group by multipolygon values, so
        # defer such fields in the aggregation query.
        for c in qs.defer('mpoly'):
            result = c.area_sum
            # If the result is a measure object, get value.
            if isinstance(result, Area):
                result = result.sq_m
            self.assertAlmostEqual((result - c.mpoly.area) / c.mpoly.area, 0)

    @skipUnlessDBFeature("has_MakeValid_function")
    def test_make_valid(self):
        invalid_geom = fromstr('POLYGON((0 0, 0 1, 1 1, 1 0, 1 1, 1 0, 0 0))')
        State.objects.create(name='invalid', poly=invalid_geom)
        invalid = State.objects.filter(name='invalid').annotate(repaired=functions.MakeValid('poly')).first()
        self.assertIs(invalid.repaired.valid, True)
        self.assertEqual(invalid.repaired, fromstr('POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))'))

    @skipUnlessDBFeature("has_MemSize_function")
    def test_memsize(self):
        ptown = City.objects.annotate(size=functions.MemSize('point')).get(name='Pueblo')
        self.assertTrue(20 <= ptown.size <= 40)  # Exact value may depend on PostGIS version

    @skipUnlessDBFeature("has_NumGeom_function")
    def test_num_geom(self):
        # Both 'countries' only have two geometries.
        for c in Country.objects.annotate(num_geom=functions.NumGeometries('mpoly')):
            self.assertEqual(2, c.num_geom)

        qs = City.objects.filter(point__isnull=False).annotate(num_geom=functions.NumGeometries('point'))
        for city in qs:
            # Oracle and PostGIS return 1 for the number of geometries on
            # non-collections, whereas MySQL returns None.
            if mysql:
                self.assertIsNone(city.num_geom)
            else:
                self.assertEqual(1, city.num_geom)

    @skipUnlessDBFeature("has_NumPoint_function")
    def test_num_points(self):
        coords = [(-95.363151, 29.763374), (-95.448601, 29.713803)]
        Track.objects.create(name='Foo', line=LineString(coords))
        qs = Track.objects.annotate(num_points=functions.NumPoints('line'))
        self.assertEqual(qs.first().num_points, 2)