/
fields.py
73 lines (58 loc) · 2.22 KB
/
fields.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
# This file is part of the Open Data Cube, see https://opendatacube.org for more information
#
# Copyright (c) 2015-2023 ODC Contributors
# SPDX-License-Identifier: Apache-2.0
"""
Common datatypes for DB drivers.
"""
from datetime import date, datetime, time
from dateutil.tz import tz
from typing import List
from datacube.model import Range
from datacube.model.fields import Expression, Field
__all__ = ['Field',
'Expression',
'OrExpression',
'UnknownFieldError',
'to_expressions',
'as_expression']
class UnknownFieldError(Exception):
pass
class OrExpression(Expression):
def __init__(self, *exprs):
super(OrExpression, self).__init__()
self.exprs = exprs
# Or expressions built by dc.load are always made up of simple expressions that share the same field.
self.field = exprs[0].field
def evaluate(self, ctx):
return any(expr.evaluate(ctx) for expr in self.exprs)
def as_expression(field: Field, value) -> Expression:
"""
Convert a single field/value to expression, following the "simple" conventions.
"""
if isinstance(value, Range):
return field.between(value.begin, value.end)
elif isinstance(value, list):
return OrExpression(*(as_expression(field, val) for val in value))
# Treat a date (day) as a time range.
elif isinstance(value, date) and not isinstance(value, datetime):
return as_expression(
field,
Range(
datetime.combine(value, time.min.replace(tzinfo=tz.tzutc())),
datetime.combine(value, time.max.replace(tzinfo=tz.tzutc()))
)
)
return field == value
def _to_expression(get_field, name: str, value) -> Expression:
field = get_field(name)
if field is None:
raise UnknownFieldError('Unknown field %r' % name)
return as_expression(field, value)
def to_expressions(get_field, **query) -> List[Expression]:
"""
Convert a simple query (dict of param names and values) to expression objects.
:type get_field: (str) -> Field
:type query: dict[str,str|float|datacube.model.Range]
"""
return [_to_expression(get_field, name, value) for name, value in query.items()]