1 # SPDX-License-Identifier: GPL-3.0-or-later
3 # This file is part of Nominatim. (https://nominatim.org)
5 # Copyright (C) 2024 by the Nominatim developer community.
6 # For a full list of authors see the git log.
8 Datastructures for a tokenized query.
10 from typing import List, Tuple, Optional, Iterator
11 from abc import ABC, abstractmethod
16 class BreakType(enum.Enum):
17 """ Type of break between tokens.
20 """ Begin of the query. """
22 """ End of the query. """
24 """ Hard break between two phrases. Address parts cannot cross hard
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.
32 """ Break between words. """
34 """ Break inside a word, for example a hyphen or apostrophe. """
36 """ Break created as a result of tokenization.
37 This may happen in languages without spaces between words.
41 class TokenType(enum.Enum):
45 """ Full name of a place. """
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. """
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 ...). """
60 class PhraseType(enum.Enum):
61 """ Designation of a phrase.
64 """ No specific designation (i.e. source is free-form query). """
66 """ Contains name or type of a POI. """
68 """ Contains a street name optionally with a housenumber. """
70 """ Contains the postal city. """
72 """ Contains the equivalent of a county. """
74 """ Contains a state or province. """
75 POSTCODE = enum.auto()
76 """ Contains a postal code. """
78 """ Contains the country name or code. """
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.
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
97 return ttype in (TokenType.WORD, TokenType.PARTIAL)
100 @dataclasses.dataclass
102 """ Base type for tokens.
103 Specific query analyzers must implement the concrete token class.
113 def get_category(self) -> Tuple[str, str]:
114 """ Return the category restriction for qualifier terms and
119 @dataclasses.dataclass
121 """ Indexes of query nodes over which a token spans.
125 penalty: Optional[float] = None
127 def __lt__(self, other: 'TokenRange') -> bool:
128 return self.end <= other.start
130 def __le__(self, other: 'TokenRange') -> bool:
131 return NotImplemented
133 def __gt__(self, other: 'TokenRange') -> bool:
134 return self.start >= other.end
136 def __ge__(self, other: 'TokenRange') -> bool:
137 return NotImplemented
139 def replace_start(self, new_start: int) -> 'TokenRange':
140 """ Return a new token range with the new start.
142 return TokenRange(new_start, self.end)
144 def replace_end(self, new_end: int) -> 'TokenRange':
145 """ Return a new token range with the new end.
147 return TokenRange(self.start, new_end)
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.
153 return self.replace_end(index), self.replace_start(index)
156 @dataclasses.dataclass
158 """ List of all tokens of a given type going from one breakpoint to another.
164 def add_penalty(self, penalty: float) -> None:
165 """ Add the given penalty to all tokens in the list.
167 for token in self.tokens:
168 token.penalty += penalty
171 @dataclasses.dataclass
173 """ A node of the query representing a break between terms.
177 starting: List[TokenList] = dataclasses.field(default_factory=list)
179 def has_tokens(self, end: int, *ttypes: TokenType) -> bool:
180 """ Check if there are tokens of the given types ending at the
183 return any(tl.end == end and tl.ttype in ttypes for tl in self.starting)
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
190 for tlist in self.starting:
191 if tlist.end == end and tlist.ttype == ttype:
196 @dataclasses.dataclass
198 """ A normalized query part. Phrases may be typed which means that
199 they then represent a specific part of the address.
206 """ A tokenized search query together with the normalized source
207 from which the tokens have been parsed.
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.
214 When created, a query contains a single node: the start of the
215 query. Further nodes can be added by appending to 'nodes'.
218 def __init__(self, source: List[Phrase]) -> None:
220 self.nodes: List[QueryNode] = \
221 [QueryNode(BreakType.START, source[0].ptype if source else PhraseType.NONE)]
223 def num_token_slots(self) -> int:
224 """ Return the length of the query in vertice steps.
226 return len(self.nodes) - 1
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
233 self.nodes.append(QueryNode(btype, ptype))
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
242 If the token type is not compatible with the phrase it should
243 be added to, then the token is silently dropped.
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)
251 snode.starting.append(TokenList(trange.end, ttype, [token]))
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.
260 return self.nodes[trange.start].get_tokens(trange.end, ttype) or []
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
268 return [next(iter(self.get_tokens(TokenRange(i, i+1), TokenType.PARTIAL)))
269 for i in range(trange.start, trange.end)]
271 def iter_token_lists(self) -> Iterator[Tuple[int, QueryNode, TokenList]]:
272 """ Iterator over all token lists in the query.
274 for i, node in enumerate(self.nodes):
275 for tlist in node.starting:
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
284 for node in self.nodes:
285 for tlist in node.starting:
286 for t in tlist.tokens:
288 return f"[{tlist.ttype.name[0]}]{t.lookup_word}"