opening-hours 2.0.2

A parser and evaluation tool for the opening_hours fields in OpenStreetMap.
Documentation 
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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

#!/usr/bin/env python3
"""
Fetch examples from taginfo.openstreetmap.org.
"""

import asyncio
import csv
from pathlib import Path

import aiohttp
from opening_hours import OpeningHours

CRATE_ROOT = Path(__file__).parent.parent
API_URL = "https://taginfo.openstreetmap.org/api/4/key/values"
PAGE_LENGTH = 999


def bool_csv_str(b: bool) -> str:
    return "yes" if b else "no"


async def main():
    count_ok = 0
    count_total = 0
    page = 1

    async with aiohttp.ClientSession() as http:
        with open(CRATE_ROOT / "opening-hours" / "data" / "osm_examples.csv", "w") as f:
            output = csv.DictWriter(
                f,
                [
                    "count",
                    "expression",
                    "parser ok",
                    "normalized",
                    "next change",
                    "error",
                    "warnings",
                ],
            )

            output.writeheader()

            while True:
                async with http.get(
                    API_URL,
                    params={
                        "key": "opening_hours",
                        "sortname": "count",
                        "sortorder": "desc",
                        "rp": PAGE_LENGTH,
                        "page": page,
                    },
                ) as resp:
                    content = await resp.json()

                for line in content["data"]:
                    oh = None
                    error = ""

                    try:
                        oh = OpeningHours(line["value"], max_interval_days=366)
                    except Exception as exc:
                        error = str(exc)

                    res = {
                        "count": line["count"],
                        "expression": line["value"],
                        "parser ok": bool_csv_str(oh is not None),
                        "normalized": "",
                        "next change": "",
                        "error": error,
                        "warnings": "",
                    }

                    if oh is not None:
                        count_ok += line["count"]
                        res["normalized"] = str(oh.normalize())
                        res["warnings"] = ",".join(oh.warnings)

                        if dt := oh.next_change():
                            res["next change"] = str(dt.isoformat())

                    output.writerow(res)
                    count_total += line["count"]

                print(f"Page {page}")
                page += 1

                if len(content["data"]) < PAGE_LENGTH:
                    break

    percent = 100 * count_ok / count_total
    print(f"Correct evaluation : {count_ok}/{count_total} ({percent:.2f}%)")


if __name__ == "__main__":
    asyncio.run(main())