forked from osm-fr/osmose-backend
-
Notifications
You must be signed in to change notification settings - Fork 0
/
analyser_merge_police_FR_gn.py
115 lines (102 loc) · 6.12 KB
/
analyser_merge_police_FR_gn.py
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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
#!/usr/bin/env python
#-*- coding: utf-8 -*-
###########################################################################
## ##
## Copyrights Frédéric Rodrigo 2014-2016 ##
## ##
## This program is free software: you can redistribute it and/or modify ##
## it under the terms of the GNU General Public License as published by ##
## the Free Software Foundation, either version 3 of the License, or ##
## (at your option) any later version. ##
## ##
## This program is distributed in the hope that it will be useful, ##
## but WITHOUT ANY WARRANTY; without even the implied warranty of ##
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the ##
## GNU General Public License for more details. ##
## ##
## You should have received a copy of the GNU General Public License ##
## along with this program. If not, see <http://www.gnu.org/licenses/>. ##
## ##
###########################################################################
from modules.OsmoseTranslation import T_
from collections import OrderedDict
from .Analyser_Merge import Analyser_Merge, Source, CSV, Load, Mapping, Select, Generate
class Analyser_Merge_Police_FR_gn(Analyser_Merge):
def __init__(self, config, logger = None):
Analyser_Merge.__init__(self, config, logger)
self.def_class_missing_official(item = 8190, id = 1, level = 3, tags = ['merge'],
title = T_('Police/"Gendarmerie" not integrated'))
self.def_class_possible_merge(item = 8191, id = 3, level = 3, tags = ['merge'],
title = T_('Police/"Gendarmerie", integration suggestion'))
self.def_class_update_official(item = 8192, id = 4, level = 3, tags = ['merge'],
title = T_('Police/"Gendarmerie" update'))
self.init(
u"https://www.data.gouv.fr/fr/datasets/liste-des-unites-de-gendarmerie-accueillant-du-public-comprenant-leur-geolocalisation-et-leurs-horaires-douverture/",
u"Liste des points d'accueil de la gendarmerie nationale",
CSV(Source(attribution = u"data.gouv.fr:Ministère de l'Intérieur", millesime = "10/2018",
fileUrl = u"https://www.data.gouv.fr/fr/datasets/r/d6a43ef2-d302-4456-90e9-ff2c47cac562"),
separator = u";"),
Load("geocodage_x_GPS", "geocodage_y_GPS",
where = lambda row: u"Centre d'information et de recrutement" not in row["service"] and u"motorisé" not in row["service"] ),
Mapping(
select = Select(
types = ["nodes", "ways"],
tags = {"amenity": "police"}),
conflationDistance = 500,
osmRef = "ref:FR:GendarmerieNationale",
generate = Generate(
static1 = {
"amenity": "police",
"name": "Gendarmerie nationale",
"police:FR": "gendarmerie",
"operator:wikidata": "Q1422336",
"operator": "Gendarmerie nationale"},
static2 = {"source": self.source},
mapping1 = {
"ref:FR:GendarmerieNationale": "identifiant_public_unite",
"opening_hours": lambda fields: parse_opening_hours(fields),
"seasonal": lambda fields: "yes" if "Poste provisoire" in fields["service"] else None},
mapping2 = {
"phone": "telephone",
"official_name": "service",
},
text = lambda tags, fields: {"en": "{0}, {1}".format(fields["service"], fields["adresse_geographique"])} )))
OSM_DAYS_FR = ['lundi', 'mardi', 'mercredi', 'jeudi', 'vendredi', 'samedi', 'dimanche', 'jours_feries']
OSM_DAYS = OrderedDict(lundi='Mo', mardi='Tu', mercredi='We', jeudi='Th', vendredi='Fr', samedi='Sa', dimanche='Su', jours_feries='PH')
def parse_opening_hours(line):
hours_list = []
for a_day in OSM_DAYS_FR:
hours = ''
if line[a_day + '_plage1_fin']:
hours += '{}-{}'.format(line[a_day + '_plage1_debut'], line[a_day + '_plage1_fin'])
if line[a_day + '_plage2_fin']:
hours += ',{}-{}'.format(line[a_day + '_plage2_debut'], line[a_day + '_plage2_fin'])
if line[a_day + '_plage3_fin']:
hours += ',{}-{}'.format(line[a_day + '_plage3_debut'], line[a_day + '_plage3_fin'])
hours_list.append({'hours': hours.replace('h', ':'), 'day': OSM_DAYS[a_day]})
hours_text = ''
day_two = 0
for i in range(7):
if i < day_two:
continue
current_hours_txt = ''
for j in range(7):
if j < i:
continue
day_two = j
if hours_list[j]['hours'] == hours_list[i]['hours']:
if hours_list[j]['hours']:
if i == j:
current_hours_txt = '{} {}; '.format(hours_list[i]['day'], hours_list[i]['hours'])
else:
current_hours_txt = '{}-{} {}; '.format(hours_list[i]['day'], hours_list[j]['day'], hours_list[i]['hours'])
else:
break
hours_text += current_hours_txt
if hours_list[7]['hours']:
hours_text += '{} {}'.format(hours_list[7]['day'], hours_list[7]['hours'])
if hours_text.endswith('; '):
hours_text = hours_text[0:-2]
if not hours_text:
hours_text = None
return hours_text