]> git.openstreetmap.org Git - nominatim.git/blob - src/nominatim_api/search/query.py
add inner word break penalty
[nominatim.git] / src / nominatim_api / search / query.py
1 # SPDX-License-Identifier: GPL-3.0-or-later
2 #
3 # This file is part of Nominatim. (https://nominatim.org)
4 #
5 # Copyright (C) 2024 by the Nominatim developer community.
6 # For a full list of authors see the git log.
7 """
8 Datastructures for a tokenized query.
9 """
10 from typing import List, Tuple, Optional, Iterator
11 from abc import ABC, abstractmethod
12 import dataclasses
13 import enum
14
15
16 class BreakType(enum.Enum):
17     """ Type of break between tokens.
18     """
19     START = '<'
20     """ Begin of the query. """
21     END = '>'
22     """ End of the query. """
23     PHRASE = ','
24     """ Hard break between two phrases. Address parts cannot cross hard
25         phrase boundaries."""
26     SOFT_PHRASE = ':'
27     """ Likely break between two phrases. Address parts should not cross soft
28         phrase boundaries. Soft breaks can be inserted by a preprocessor
29         that is analysing the input string.
30     """
31     WORD = ' '
32     """ Break between words. """
33     PART = '-'
34     """ Break inside a word, for example a hyphen or apostrophe. """
35     TOKEN = '`'
36     """ Break created as a result of tokenization.
37         This may happen in languages without spaces between words.
38     """
39
40
41 class TokenType(enum.Enum):
42     """ Type of token.
43     """
44     WORD = enum.auto()
45     """ Full name of a place. """
46     PARTIAL = enum.auto()
47     """ Word term without breaks, does not necessarily represent a full name. """
48     HOUSENUMBER = enum.auto()
49     """ Housenumber term. """
50     POSTCODE = enum.auto()
51     """ Postal code term. """
52     COUNTRY = enum.auto()
53     """ Country name or reference. """
54     QUALIFIER = enum.auto()
55     """ Special term used together with name (e.g. _Hotel_ Bellevue). """
56     NEAR_ITEM = enum.auto()
57     """ Special term used as searchable object(e.g. supermarket in ...). """
58
59
60 class PhraseType(enum.Enum):
61     """ Designation of a phrase.
62     """
63     NONE = 0
64     """ No specific designation (i.e. source is free-form query). """
65     AMENITY = enum.auto()
66     """ Contains name or type of a POI. """
67     STREET = enum.auto()
68     """ Contains a street name optionally with a housenumber. """
69     CITY = enum.auto()
70     """ Contains the postal city. """
71     COUNTY = enum.auto()
72     """ Contains the equivalent of a county. """
73     STATE = enum.auto()
74     """ Contains a state or province. """
75     POSTCODE = enum.auto()
76     """ Contains a postal code. """
77     COUNTRY = enum.auto()
78     """ Contains the country name or code. """
79
80     def compatible_with(self, ttype: TokenType,
81                         is_full_phrase: bool) -> bool:
82         """ Check if the given token type can be used with the phrase type.
83         """
84         if self == PhraseType.NONE:
85             return not is_full_phrase or ttype != TokenType.QUALIFIER
86         if self == PhraseType.AMENITY:
87             return ttype in (TokenType.WORD, TokenType.PARTIAL)\
88                    or (is_full_phrase and ttype == TokenType.NEAR_ITEM)\
89                    or (not is_full_phrase and ttype == TokenType.QUALIFIER)
90         if self == PhraseType.STREET:
91             return ttype in (TokenType.WORD, TokenType.PARTIAL, TokenType.HOUSENUMBER)
92         if self == PhraseType.POSTCODE:
93             return ttype == TokenType.POSTCODE
94         if self == PhraseType.COUNTRY:
95             return ttype == TokenType.COUNTRY
96
97         return ttype in (TokenType.WORD, TokenType.PARTIAL)
98
99
100 @dataclasses.dataclass
101 class Token(ABC):
102     """ Base type for tokens.
103         Specific query analyzers must implement the concrete token class.
104     """
105
106     penalty: float
107     token: int
108     count: int
109     addr_count: int
110     lookup_word: str
111
112     @abstractmethod
113     def get_category(self) -> Tuple[str, str]:
114         """ Return the category restriction for qualifier terms and
115             category objects.
116         """
117
118
119 @dataclasses.dataclass
120 class TokenRange:
121     """ Indexes of query nodes over which a token spans.
122     """
123     start: int
124     end: int
125     penalty: Optional[float] = None
126
127     def __lt__(self, other: 'TokenRange') -> bool:
128         return self.end <= other.start
129
130     def __le__(self, other: 'TokenRange') -> bool:
131         return NotImplemented
132
133     def __gt__(self, other: 'TokenRange') -> bool:
134         return self.start >= other.end
135
136     def __ge__(self, other: 'TokenRange') -> bool:
137         return NotImplemented
138
139     def replace_start(self, new_start: int) -> 'TokenRange':
140         """ Return a new token range with the new start.
141         """
142         return TokenRange(new_start, self.end)
143
144     def replace_end(self, new_end: int) -> 'TokenRange':
145         """ Return a new token range with the new end.
146         """
147         return TokenRange(self.start, new_end)
148
149     def split(self, index: int) -> Tuple['TokenRange', 'TokenRange']:
150         """ Split the span into two spans at the given index.
151             The index must be within the span.
152         """
153         return self.replace_end(index), self.replace_start(index)
154
155
156 @dataclasses.dataclass
157 class TokenList:
158     """ List of all tokens of a given type going from one breakpoint to another.
159     """
160     end: int
161     ttype: TokenType
162     tokens: List[Token]
163
164     def add_penalty(self, penalty: float) -> None:
165         """ Add the given penalty to all tokens in the list.
166         """
167         for token in self.tokens:
168             token.penalty += penalty
169
170
171 @dataclasses.dataclass
172 class QueryNode:
173     """ A node of the query representing a break between terms.
174     """
175     btype: BreakType
176     ptype: PhraseType
177     starting: List[TokenList] = dataclasses.field(default_factory=list)
178
179     def has_tokens(self, end: int, *ttypes: TokenType) -> bool:
180         """ Check if there are tokens of the given types ending at the
181             given node.
182         """
183         return any(tl.end == end and tl.ttype in ttypes for tl in self.starting)
184
185     def get_tokens(self, end: int, ttype: TokenType) -> Optional[List[Token]]:
186         """ Get the list of tokens of the given type starting at this node
187             and ending at the node 'end'. Returns 'None' if no such
188             tokens exist.
189         """
190         for tlist in self.starting:
191             if tlist.end == end and tlist.ttype == ttype:
192                 return tlist.tokens
193         return None
194
195
196 @dataclasses.dataclass
197 class Phrase:
198     """ A normalized query part. Phrases may be typed which means that
199         they then represent a specific part of the address.
200     """
201     ptype: PhraseType
202     text: str
203
204
205 class QueryStruct:
206     """ A tokenized search query together with the normalized source
207         from which the tokens have been parsed.
208
209         The query contains a list of nodes that represent the breaks
210         between words. Tokens span between nodes, which don't necessarily
211         need to be direct neighbours. Thus the query is represented as a
212         directed acyclic graph.
213
214         When created, a query contains a single node: the start of the
215         query. Further nodes can be added by appending to 'nodes'.
216     """
217
218     def __init__(self, source: List[Phrase]) -> None:
219         self.source = source
220         self.nodes: List[QueryNode] = \
221             [QueryNode(BreakType.START, source[0].ptype if source else PhraseType.NONE)]
222
223     def num_token_slots(self) -> int:
224         """ Return the length of the query in vertice steps.
225         """
226         return len(self.nodes) - 1
227
228     def add_node(self, btype: BreakType, ptype: PhraseType) -> None:
229         """ Append a new break node with the given break type.
230             The phrase type denotes the type for any tokens starting
231             at the node.
232         """
233         self.nodes.append(QueryNode(btype, ptype))
234
235     def add_token(self, trange: TokenRange, ttype: TokenType, token: Token) -> None:
236         """ Add a token to the query. 'start' and 'end' are the indexes of the
237             nodes from which to which the token spans. The indexes must exist
238             and are expected to be in the same phrase.
239             'ttype' denotes the type of the token and 'token' the token to
240             be inserted.
241
242             If the token type is not compatible with the phrase it should
243             be added to, then the token is silently dropped.
244         """
245         snode = self.nodes[trange.start]
246         full_phrase = snode.btype in (BreakType.START, BreakType.PHRASE)\
247             and self.nodes[trange.end].btype in (BreakType.PHRASE, BreakType.END)
248         if snode.ptype.compatible_with(ttype, full_phrase):
249             tlist = snode.get_tokens(trange.end, ttype)
250             if tlist is None:
251                 snode.starting.append(TokenList(trange.end, ttype, [token]))
252             else:
253                 tlist.append(token)
254
255     def get_tokens(self, trange: TokenRange, ttype: TokenType) -> List[Token]:
256         """ Get the list of tokens of a given type, spanning the given
257             nodes. The nodes must exist. If no tokens exist, an
258             empty list is returned.
259         """
260         return self.nodes[trange.start].get_tokens(trange.end, ttype) or []
261
262     def get_partials_list(self, trange: TokenRange) -> List[Token]:
263         """ Create a list of partial tokens between the given nodes.
264             The list is composed of the first token of type PARTIAL
265             going to the subsequent node. Such PARTIAL tokens are
266             assumed to exist.
267         """
268         return [next(iter(self.get_tokens(TokenRange(i, i+1), TokenType.PARTIAL)))
269                 for i in range(trange.start, trange.end)]
270
271     def iter_token_lists(self) -> Iterator[Tuple[int, QueryNode, TokenList]]:
272         """ Iterator over all token lists in the query.
273         """
274         for i, node in enumerate(self.nodes):
275             for tlist in node.starting:
276                 yield i, node, tlist
277
278     def find_lookup_word_by_id(self, token: int) -> str:
279         """ Find the first token with the given token ID and return
280             its lookup word. Returns 'None' if no such token exists.
281             The function is very slow and must only be used for
282             debugging.
283         """
284         for node in self.nodes:
285             for tlist in node.starting:
286                 for t in tlist.tokens:
287                     if t.token == token:
288                         return f"[{tlist.ttype.name[0]}]{t.lookup_word}"
289         return 'None'