def find_all_tiles_in_polygon()

in data-access/nexustiles/dao/ElasticsearchProxy.py [0:0]

• 51 lines of code
• 4 McCabe index (conditional complexity)

    def find_all_tiles_in_polygon(self, bounding_polygon, ds, start_time=0, end_time=-1, **kwargs):

        nums = re.findall(r'\d+(?:\.\d*)?', bounding_polygon.wkt.rpartition(',')[0])
        polygon_coordinates = list(zip(*[iter(nums)] * 2))

        max_lat = bounding_polygon.bounds[3]
        min_lon = bounding_polygon.bounds[0]
        min_lat = bounding_polygon.bounds[1]
        max_lon = bounding_polygon.bounds[2]

        params = {
            "size": 1000,
            "query": {
                "bool": {
                    "filter": [
                        { 
                            "term": {
                                "dataset_s": {
                                    "value": ds
                                }
                            } 
                        },
                        {
                            "geo_shape": {
                                "geo": {
                                    "shape": {
                                        "type": "envelope",
                                        "coordinates": [[min_lon, max_lat], [max_lon, min_lat]]
                                    },
                                    "relation": "intersects"
                                }
                            }
                        },
                        {
                            "range": {
                                "tile_count_i": {
                                    "gte": 1
                                }
                            }
                        }
                    ]
                }
            }
        }

        try:
            if 'fl' in list(kwargs.keys()):
                params["_source"] = kwargs["fl"].split(',')
        except KeyError:
            pass

        if 0 < start_time <= end_time:
            params["query"]["bool"]["should"] = self.get_formatted_time_clause(start_time, end_time)
            params["query"]["bool"]["minimum_should_match"] = 1

        self._merge_kwargs(params, **kwargs)

        return self.do_query_all(*(None, None, None, False, None), **params)