278 lines
11 KiB
Python
278 lines
11 KiB
Python
import json
|
|
from pathlib import Path
|
|
from django.core.management.base import BaseCommand
|
|
from django.contrib.gis.geos import Point, Polygon, MultiPolygon
|
|
from django.db import transaction
|
|
from shapely.geometry import shape, Point as ShpPoint
|
|
|
|
from common.models import Thematic, Municipality
|
|
from assets.models import AssetCategory
|
|
from assets.models.sign import SignStreet
|
|
from assets.models.parking import ParkingLocation, ParkingSpot
|
|
|
|
class Command(BaseCommand):
|
|
help = "Import parking areas, bicycle parking stands and PMR/VE spots for Court-Saint-Étienne"
|
|
|
|
def handle(self, *args, **options):
|
|
self.stdout.write("=== Importing Parking & Bicycle Spots for Court-Saint-Étienne ===")
|
|
|
|
parking_thematic = Thematic.objects.filter(code='parking').first()
|
|
parking_cat = AssetCategory.objects.filter(code='parking_spot').first()
|
|
bike_cat = AssetCategory.objects.filter(code='bike_parking_spot').first()
|
|
|
|
cse = Municipality.objects.get(niscode='25014')
|
|
cse_geom_wgs84 = cse.geom.clone()
|
|
cse_geom_wgs84.transform(4326)
|
|
cse_poly = shape(json.loads(cse_geom_wgs84.geojson))
|
|
|
|
streets = list(SignStreet.objects.filter(geom__isnull=False))
|
|
if not streets:
|
|
self.stderr.write("No SignStreet found in database!")
|
|
return
|
|
|
|
osm_json_path = Path("/tmp/osm_cse_sign_parking.json")
|
|
if not osm_json_path.exists():
|
|
self.stderr.write(f"OSM JSON not found: {osm_json_path}")
|
|
return
|
|
|
|
with open(osm_json_path, 'r', encoding='utf-8') as f:
|
|
osm_data = json.load(f)
|
|
|
|
elements = osm_data.get('elements', [])
|
|
|
|
# Filter strictly inside CSE
|
|
in_cse = []
|
|
for el in elements:
|
|
lat = el.get('lat') or (el.get('center') and el['center'].get('lat'))
|
|
lon = el.get('lon') or (el.get('center') and el['center'].get('lon'))
|
|
if lat is not None and lon is not None:
|
|
p = ShpPoint(lon, lat)
|
|
if cse_poly.contains(p):
|
|
el['_pt_wgs84'] = Point(lon, lat, srid=4326)
|
|
in_cse.append(el)
|
|
|
|
self.stdout.write(f"Elements strictly in CSE: {len(in_cse)}")
|
|
|
|
# ─── 1. ParkingLocation (Zones de stationnement) ───────────────────
|
|
self.stdout.write("--- Step 1: ParkingLocation (Aires/Parcs) ---")
|
|
parking_elements = [e for e in in_cse if e.get('tags', {}).get('amenity') == 'parking']
|
|
self.stdout.write(f"Parking lots in CSE: {len(parking_elements)}")
|
|
|
|
# Existing locations
|
|
existing_loc_codes = set(ParkingLocation.objects.values_list('code', flat=True))
|
|
max_loc_num = 0
|
|
for c in existing_loc_codes:
|
|
if c and c.startswith('CSE-PKG-'):
|
|
try:
|
|
num = int(c.split('-')[-1])
|
|
if num > max_loc_num:
|
|
max_loc_num = num
|
|
except ValueError:
|
|
pass
|
|
|
|
new_locations = []
|
|
for p in parking_elements:
|
|
tags = p.get('tags', {})
|
|
pt_4326 = p['_pt_wgs84']
|
|
pt_3812 = pt_4326.transform(3812, clone=True)
|
|
|
|
closest_street = min(streets, key=lambda st: st.geom.distance(pt_3812))
|
|
|
|
# Determine Name
|
|
osm_name = tags.get('name')
|
|
if osm_name:
|
|
name_fr = osm_name.strip()
|
|
else:
|
|
name_fr = f"Parking {closest_street.name_fr or 'public'}"
|
|
|
|
# Capacity
|
|
try:
|
|
cap = int(tags.get('capacity', 15))
|
|
except (ValueError, TypeError):
|
|
cap = 15
|
|
|
|
# Size proportional to capacity
|
|
size = max(16.0, min(65.0, (cap or 15) * 1.8))
|
|
half = size / 2.0
|
|
cx, cy = pt_3812.x, pt_3812.y
|
|
poly = Polygon([
|
|
(cx - half, cy - half),
|
|
(cx + half, cy - half),
|
|
(cx + half, cy + half),
|
|
(cx - half, cy + half),
|
|
(cx - half, cy - half),
|
|
], srid=3812)
|
|
mpoly = MultiPolygon([poly])
|
|
|
|
access = tags.get('access', 'public')
|
|
fee = tags.get('fee', 'non payant')
|
|
parking_type = tags.get('parking', 'surface')
|
|
desc = f"Aire de stationnement ({parking_type}). Accès: {access}. Tarification: {fee}. Capacité estimée: {cap} places."
|
|
|
|
max_loc_num += 1
|
|
code = f"CSE-PKG-{max_loc_num:04d}"
|
|
|
|
loc = ParkingLocation(
|
|
code=code,
|
|
name_fr=name_fr,
|
|
geom=mpoly,
|
|
description=desc,
|
|
address=f"{closest_street.name_fr or 'Court-Saint-Étienne'}, 1490 Court-Saint-Étienne",
|
|
|
|
)
|
|
# Sync
|
|
loc.sync_geom_and_geojson()
|
|
new_locations.append(loc)
|
|
|
|
with transaction.atomic():
|
|
ParkingLocation.objects.bulk_create(new_locations)
|
|
self.stdout.write(self.style.SUCCESS(f"Created {len(new_locations)} ParkingLocations. Total in DB: {ParkingLocation.objects.count()}"))
|
|
|
|
# Re-fetch all locations with spatial index
|
|
all_locations = list(ParkingLocation.objects.filter(geom__isnull=False))
|
|
|
|
# ─── 2. ParkingSpot (Vélos & Emplacements individuels) ────────────
|
|
self.stdout.write("--- Step 2: ParkingSpot (Vélos, PMR, VE) ---")
|
|
bike_elements = [e for e in in_cse if e.get('tags', {}).get('amenity') == 'bicycle_parking']
|
|
space_elements = [e for e in in_cse if e.get('tags', {}).get('amenity') == 'parking_space']
|
|
|
|
self.stdout.write(f"Bicycle parkings in CSE: {len(bike_elements)}")
|
|
self.stdout.write(f"Parking spaces in CSE: {len(space_elements)}")
|
|
|
|
max_spot_num = 0
|
|
for sc in ParkingSpot.objects.values_list('code', flat=True):
|
|
if sc:
|
|
try:
|
|
num = int(sc.split('-')[-1])
|
|
if num > max_spot_num:
|
|
max_spot_num = num
|
|
except ValueError:
|
|
pass
|
|
|
|
new_spots = []
|
|
|
|
# A. Bicycle parkings
|
|
for b in bike_elements:
|
|
tags = b.get('tags', {})
|
|
pt_4326 = b['_pt_wgs84']
|
|
pt_3812 = pt_4326.transform(3812, clone=True)
|
|
|
|
closest_street = min(streets, key=lambda st: st.geom.distance(pt_3812))
|
|
closest_loc = min(all_locations, key=lambda l: l.geom.distance(pt_3812)) if all_locations else None
|
|
|
|
# Capacity & stands
|
|
try:
|
|
cap = int(tags.get('capacity', 6))
|
|
except (ValueError, TypeError):
|
|
cap = 6
|
|
|
|
btype_osm = (tags.get('bicycle_parking') or '').lower()
|
|
if 'box' in btype_osm or 'shed' in btype_osm or 'secure' in btype_osm:
|
|
stand_type = 'box'
|
|
stands_label = "Box / Abri sécurisé"
|
|
elif 'rack' in btype_osm or 'wall' in btype_osm:
|
|
stand_type = 'rack'
|
|
stands_label = "Râteau / Pince-roue"
|
|
else:
|
|
stand_type = 'normal'
|
|
stands_label = "Arceaux U standard"
|
|
|
|
# 3m x 2m polygon in 3812
|
|
cx, cy = pt_3812.x, pt_3812.y
|
|
poly = Polygon([
|
|
(cx - 1.5, cy - 1.0),
|
|
(cx + 1.5, cy - 1.0),
|
|
(cx + 1.5, cy + 1.0),
|
|
(cx - 1.5, cy + 1.0),
|
|
(cx - 1.5, cy - 1.0),
|
|
], srid=3812)
|
|
|
|
max_spot_num += 1
|
|
code = f"CSE-PKV-{max_spot_num:04d}"
|
|
osm_name = tags.get('name')
|
|
name_fr = osm_name if osm_name else f"Stationnement vélos ({stands_label}) - {closest_street.name_fr or 'Rue'}"
|
|
|
|
spot = ParkingSpot(
|
|
code=code,
|
|
name_fr=name_fr,
|
|
spot_type='bicycle',
|
|
stand_type=stand_type,
|
|
capacity=cap,
|
|
standard_stand_count=max(1, cap // 2),
|
|
geom=poly,
|
|
location=closest_loc,
|
|
street=closest_street,
|
|
|
|
category=bike_cat,
|
|
|
|
is_compliant=True,
|
|
)
|
|
spot.sync_geom_and_geojson()
|
|
new_spots.append(spot)
|
|
|
|
# B. Parking Spaces (PMR, EV charging, Car)
|
|
for s in space_elements:
|
|
tags = s.get('tags', {})
|
|
pt_4326 = s['_pt_wgs84']
|
|
pt_3812 = pt_4326.transform(3812, clone=True)
|
|
|
|
closest_street = min(streets, key=lambda st: st.geom.distance(pt_3812))
|
|
closest_loc = min(all_locations, key=lambda l: l.geom.distance(pt_3812)) if all_locations else None
|
|
|
|
# Check PMR vs EV vs standard
|
|
is_pmr = (
|
|
tags.get('capacity:disabled') in ('1', 'yes') or
|
|
tags.get('disabled') in ('yes', 'designated') or
|
|
tags.get('access') == 'disabled' or
|
|
'pmr' in str(tags).lower() or
|
|
'disabled' in str(tags).lower()
|
|
)
|
|
is_ev = (
|
|
'charging' in str(tags).lower() or
|
|
tags.get('socket:type2') or
|
|
tags.get('amenity') == 'charging_station'
|
|
)
|
|
|
|
if is_pmr:
|
|
spot_type = 'pmr'
|
|
spot_label = "Emplacement PMR"
|
|
elif is_ev:
|
|
spot_type = 'ev_charging'
|
|
spot_label = "Borne recharge VE"
|
|
else:
|
|
spot_type = 'car_parking'
|
|
spot_label = "Emplacement automobile"
|
|
|
|
# 2.5m x 5m polygon
|
|
cx, cy = pt_3812.x, pt_3812.y
|
|
poly = Polygon([
|
|
(cx - 1.25, cy - 2.5),
|
|
(cx + 1.25, cy - 2.5),
|
|
(cx + 1.25, cy + 2.5),
|
|
(cx - 1.25, cy + 2.5),
|
|
(cx - 1.25, cy - 2.5),
|
|
], srid=3812)
|
|
|
|
max_spot_num += 1
|
|
code = f"CSE-PKS-{max_spot_num:04d}"
|
|
name_fr = f"{spot_label} - {closest_street.name_fr or 'Rue'}"
|
|
|
|
spot = ParkingSpot(
|
|
code=code,
|
|
name_fr=name_fr,
|
|
spot_type=spot_type,
|
|
capacity=1,
|
|
geom=poly,
|
|
location=closest_loc,
|
|
street=closest_street,
|
|
|
|
category=parking_cat,
|
|
|
|
is_compliant=True,
|
|
)
|
|
spot.sync_geom_and_geojson()
|
|
new_spots.append(spot)
|
|
|
|
with transaction.atomic():
|
|
ParkingSpot.objects.bulk_create(new_spots)
|
|
self.stdout.write(self.style.SUCCESS(f"Created {len(new_spots)} ParkingSpots (total: {ParkingSpot.objects.count()})."))
|