Coverage for scripts/scratch.py: 4%
35 statements
« prev ^ index » next coverage.py v7.6.1, created at 2025-07-22 06:43 -0400
« prev ^ index » next coverage.py v7.6.1, created at 2025-07-22 06:43 -0400
1import timeit
2from src.overturetoosm.resources import places_tags
3from src.overturetoosm.places import process_place
6sample = {
7 "id": "123",
8 "version": 1,
9 "update_time": "2022-01-01T00:00:00Z",
10 "sources": [
11 {
12 "property": "property1",
13 "dataset": "dataset1",
14 "record_id": "record1",
15 "confidence": 0.8,
16 }
17 ],
18 "names": {
19 "primary": "Primary Name",
20 "common": "Common Name",
21 "rules": "Rules Name",
22 },
23 "brand": {
24 "wikidata": "Q123",
25 "names": {
26 "primary": "Brand Name",
27 "common": "Common Name",
28 "rules": "Rules Name",
29 },
30 },
31 "categories": {
32 "main": "notary_public",
33 "alternate": ["alternate_category1", "alternate_category2"],
34 },
35 "confidence": 0.8,
36 "websites": ["https://example.com"],
37 "socials": ["www.facebook.com/example"],
38 "phones": ["+1234567890"],
39 "addresses": [
40 {
41 "freeform": "123 Main St",
42 "locality": "City",
43 "postcode": "12345",
44 "region": "State",
45 "country": "Country",
46 }
47 ],
48}
50after = {
51 "names.primary": "Primary Name",
52 "brand.wikidata": "Q123",
53 "brand.names.primary": "Brand Name",
54 "confidence": 0.8,
55 "websites": "https://example.com",
56 "socials": ["www.facebook.com/example"],
57 "phones": "+1234567890",
58 "freeform": "123 Main St",
59 "locality": "City",
60 "postcode": "12345",
61 "region": "State",
62 "country": "Country",
63 "source": "dataset1 via overturetoosm",
64 "office": "lawyer",
65 "lawyer": "notary",
66}
69mapping = {
70 "names.primary": "name",
71 "brand.wikidata": "brand:wikidata",
72 "brand.names.primary": "brand",
73 "phones": "phone",
74 "freeform": "addr:street_address",
75 "locality": "addr:city",
76 "postcode": "addr:postcode",
77 "region": "addr:state",
78 "country": "addr:country",
79 "websites": "website",
80}
83def flatten(dictionary: dict, parent_key: str = "", separator: str = ".") -> dict:
84 """Flatten a nested dictionary into a single-level dictionary."""
85 items = []
86 for key, value in dictionary.items():
87 new_key = parent_key + separator + key if parent_key else key
88 if isinstance(value, dict):
89 items.extend(flatten(value, new_key, separator=separator).items())
90 elif not new_key.endswith(("rules", "common")):
91 items.append((new_key, value))
92 return dict(items)
95def get_first_item(props: dict, keys: list[str]):
96 for key in keys:
97 props[key] = props.get(key, [])[0]
98 return props
101def handle(props: dict) -> dict:
102 flattened = flatten(props)
103 x = flattened | flatten(
104 flattened.get("addresses", [])[0], parent_key="addr", separator=":"
105 )
106 x["source"] = (
107 ", ".join(source["dataset"] for source in x.get("sources", []))
108 + " via overturetoosm"
109 )
110 x = get_first_item(x, ["websites", "phones"])
111 x = x | places_tags.get(x.get("categories.main", ""), {})
112 for social in x.get("socials", []):
113 if "facebook" in social:
114 x["contact:facebook"] = social
115 elif "twitter" in social:
116 x["contact:twitter"] = social
117 remove = [
118 "update_time",
119 "id",
120 "version",
121 "addresses",
122 "sources",
123 "categories.main",
124 "categories.alternate",
125 "socials",
126 "confidence",
127 ]
128 return {mapping.get(k, k): v for k, v in x.items() if k not in remove}
131if 0:
132 a = timeit.timeit(
133 "handle(sample)", setup="from __main__ import sample, handle", number=100000
134 )
135 b = timeit.timeit(
136 "process_props(sample)",
137 setup="from __main__ import sample, process_props",
138 number=100000,
139 )
140 print("here:", a)
141 print("finished:", b)
142else:
143 print(sample)
144 print(process_place(sample))