251 lines
10 KiB
Python
251 lines
10 KiB
Python
"""Search in a parts list."""
|
|
import argparse
|
|
import asyncio
|
|
import itertools
|
|
import re
|
|
import time
|
|
from typing import Optional, Iterable
|
|
import nicegui
|
|
from nicegui import ui
|
|
from part import Part
|
|
import part_list_io
|
|
import datacleanup
|
|
|
|
|
|
def antilen(string: str):
|
|
"""Return a bigger number for shorter strings, except 0 for ""."""
|
|
return 1 / len(string) if string else 0
|
|
|
|
async def find_parts(parts: list[Part], search_string: str, max_number: int=10) -> list[Part]:
|
|
"""Gives parts that the user might have searched for.
|
|
|
|
Args:
|
|
parts: the list of parts to search in
|
|
search_string: Input of user
|
|
max_number: maximum number of returned parts
|
|
Returns:
|
|
list of parts that somehow include the search string
|
|
"""
|
|
fuzzy = re.compile(".*" + ".*".join(search_string) + ".*", flags=re.IGNORECASE)
|
|
|
|
def match_score(part):
|
|
"""Return sortable tuple of decreasing importance.
|
|
|
|
Good matches have high numbers.
|
|
"""
|
|
alt_names = itertools.chain(part.name_alt_de if hasattr(part, "name_alt_de") else [],
|
|
part.name_alt_en if hasattr(part, "name_alt_en") else [])
|
|
return (
|
|
# first prio: name at begin. Prefer short
|
|
antilen(part.name) if part.name.startswith(search_string) else 0,
|
|
# second prio: part.name_en and name_de at begin. Prefer short
|
|
max(1 / len(part.name_en)
|
|
if hasattr(part, "name_en") and part.name_en.startswith(search_string)
|
|
else 0,
|
|
1 / len(part.name_de)
|
|
if hasattr(part, "name_de") and part.name_de.startswith(search_string)
|
|
else 0),
|
|
# third prio: name with all letters appearing with gaps in correct order ("fuzzy match")
|
|
bool(fuzzy.match(part.name)),
|
|
# forth prio: name_en and name_de fuzzy match
|
|
(hasattr(part, "name_en") and bool(fuzzy.match(part.name_en)))
|
|
or (hasattr(part, "name_de") and bool(fuzzy.match(part.name_de))),
|
|
# fith prio: alternative name at begin
|
|
max((1 / len(alt_name) for alt_name in alt_names if alt_name.startswith(search_string)),
|
|
default=0),
|
|
# sixth prio: alternative name fuzzy match
|
|
max(map(antilen, filter(fuzzy.match, alt_names)),
|
|
default=0),
|
|
# seventh prio: description
|
|
max(int(hasattr(part, "description_de") and search_string in part.description_de) * 3,
|
|
int(hasattr(part, "description_de") and bool(fuzzy.match(part.description_de))),
|
|
int(hasattr(part, "description_en") and search_string in part.description_en) * 3,
|
|
int(hasattr(part, "description_en") and bool(fuzzy.match(part.description_en)))
|
|
)
|
|
)
|
|
if search_string:
|
|
scored_parts = [(part, match_score(part)) for part in parts]
|
|
return map(lambda pair: pair[0],
|
|
sorted(
|
|
filter(lambda pair: any(pair[1]),
|
|
scored_parts),
|
|
key=lambda pair: pair[1],
|
|
reverse=True
|
|
)[:max_number])
|
|
return []
|
|
|
|
|
|
def get_file_names() -> argparse.Namespace:
|
|
"""Abuse argparse for collecting file names."""
|
|
parser = argparse.ArgumentParser()
|
|
part_list_io.add_file_args(parser)
|
|
return parser.parse_args([])
|
|
|
|
|
|
async def list_parts(ui_element: nicegui.ui.element, parts: Iterable[Part]) -> None:
|
|
"""Replaces content of ui_element with information about the parts.
|
|
|
|
Args:
|
|
ui_element: Some UI element that can be changed.
|
|
"""
|
|
# gives other searches 10 ms time to abort this display which might take long
|
|
await asyncio.sleep(0.01)
|
|
ui_element.clear()
|
|
with ((ui_element)):
|
|
for part in parts:
|
|
card = ui.card()
|
|
# supplying card and part as default arguments makes it use the current
|
|
# value instead of the value at the time of usage
|
|
change_card = lambda event, c=card, p=part: show_part_changer(c, p)
|
|
with card:
|
|
print(f"Create card {id(card)} for {part.name}.")
|
|
with ui.row(wrap=False):
|
|
with ui.column():
|
|
with ui.row():
|
|
ui.label(text=part.name)
|
|
try:
|
|
name_en = part.name_en
|
|
except AttributeError:
|
|
pass
|
|
else:
|
|
if name_en != part.name:
|
|
ui.label(text=f"({name_en})").style('font-size: 70%')
|
|
ui.button("🖉").on_click(change_card)
|
|
other_names = ", ".join(itertools.chain(
|
|
vars(part).get("name_alt_de", []),
|
|
vars(part).get("name_alt_en", [])))
|
|
if other_names:
|
|
ui.label(other_names).style('font-size: 70%')
|
|
for member in ("description_de", "description_en", ""):
|
|
if hasattr(part, member):
|
|
ui.markdown(part.get(member)).style('font-size: 70%')
|
|
if part.where:
|
|
with ui.label(part.where):
|
|
ui.tooltip(part.location.long_name)
|
|
if hasattr(part, "image"):
|
|
ui.image("/images_landscape/" + part.image).props("width=50%").props(
|
|
"height=100px").props("fit='scale-down'")
|
|
|
|
|
|
def load_data() -> list[Part]:
|
|
"""Load data from text files.
|
|
|
|
Could implement that data is only loaded when necessary.
|
|
Then an argument force would be useful to reload.
|
|
|
|
Returns:
|
|
list of all things listed in the files
|
|
"""
|
|
options = get_file_names()
|
|
return part_list_io.get_parts(options)
|
|
|
|
|
|
def save_data(parts) -> None:
|
|
"""Save parts to files."""
|
|
options = get_file_names()
|
|
part_list_io.save_parts(parts, options.input_file, options.locations_file)
|
|
|
|
def search_page(parts: list[Part]) -> None:
|
|
"""Create a NiceGUI page with a search input field and search results.
|
|
|
|
Args:
|
|
parts: list of parts to search in
|
|
"""
|
|
print("(Re)build search page.")
|
|
# UI container for the search results.
|
|
results: Optional[ui.element] = None
|
|
|
|
# Search queries (max. 1) running. Here to be cancellable by different search coroutines.
|
|
running_queries: list[asyncio.Task] = []
|
|
|
|
# should use the parts as they are when clicked
|
|
ui.button("Save").on_click(lambda click_event_arguments, parts=parts: save_data(parts))
|
|
|
|
async def search(event: nicegui.events.ValueChangeEventArguments) -> None:
|
|
"""Search for cocktails as you type.
|
|
|
|
Args:
|
|
event: the input field change event. The new value event.value is used.
|
|
"""
|
|
if running_queries:
|
|
for query in running_queries:
|
|
query.cancel()
|
|
sleep = asyncio.create_task(asyncio.sleep(.5))
|
|
running_queries.append(sleep)
|
|
try:
|
|
await sleep
|
|
except asyncio.exceptions.CancelledError:
|
|
# the next letter was already typed, do not search and rerender for this query
|
|
return
|
|
query = asyncio.create_task(find_parts(parts, event.value))
|
|
running_queries.append(query)
|
|
try:
|
|
start = time.monotonic()
|
|
response = await query
|
|
end = time.monotonic()
|
|
if end - start > 0.01:
|
|
print(f"Query {event.value}: {end - start} seconds")
|
|
except asyncio.exceptions.CancelledError:
|
|
print("Search cancelled")
|
|
pass
|
|
else:
|
|
if results:
|
|
display = asyncio.create_task(list_parts(results, response))
|
|
running_queries.append(display)
|
|
try:
|
|
start = time.monotonic()
|
|
await display
|
|
if end - start > 0.01:
|
|
print(f"Display {event.value}: {end - start} seconds")
|
|
except asyncio.exceptions.CancelledError:
|
|
pass
|
|
else:
|
|
ui.notify("Internal error: results element is None.")
|
|
|
|
ui.input(on_change=search) \
|
|
.props('autofocus outlined rounded item-aligned input-class="ml-3"') \
|
|
.classes('w-96 self-center mt-24 transition-all')
|
|
results = ui.column()
|
|
|
|
|
|
|
|
def show_part_changer(ui_element: nicegui.ui.element, part: Part,
|
|
save_function: Callable[[], None]) -> None:
|
|
"""Clear content of ui element and instead display editing fields.
|
|
|
|
Args:
|
|
ui_element: the ui element (e.g. a card) on which to show the part changing ui
|
|
part: the part to change
|
|
save_function: what to call when changing is done
|
|
"""
|
|
def delete_member(member, updated: nicegui.ui.input):
|
|
del vars(part)[member]
|
|
updated.set_value("")
|
|
input_fields = {}
|
|
def save_value(event, member):
|
|
"""Copy input field value to part member."""
|
|
if not event.value:
|
|
del(vars(part)[member])
|
|
else:
|
|
vars(part)[member] = event.value
|
|
|
|
print(f"Try to let edit {part.name} with {id(ui_element)}.")
|
|
ui_element.clear()
|
|
with ui_element:
|
|
for member in ("name_de", "name_en", "description_de", "description_en"):
|
|
with ui.row():
|
|
ui.label(member + ":")
|
|
input_fields[member] = ui.input(part.get(member, "")).on_value_change(
|
|
lambda e, m=member: save_value(e, m))
|
|
ui.button("❌").on_click(lambda m=member, i=input_fields[member]: delete_member(m, i))
|
|
ui.button("Save").on_click(save_function)
|
|
|
|
|
|
if __name__ in {"__main__", "__mp_main__"}:
|
|
nicegui.app.add_static_files('/images_landscape', 'images_landscape')
|
|
search_page(load_data())
|
|
ui.run(title="Fahrradteile",
|
|
favicon="website_resources/favicon.ico",
|
|
language="de",
|
|
# host="192.168.178.133",
|
|
port=11111)
|