scaffolded PySwiss microservice: Django 6.0 project at pyswiss/ (core/ settings, apps/charts/), GET /api/chart/ + GET /api/charts/ views, EphemerisSnapshot model + migration, populate_ephemeris management command, 41 ITs (integrated + unit, all green); separate .venv with pyswisseph 2.10.3.2
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
110
pyswiss/apps/charts/views.py
Normal file
110
pyswiss/apps/charts/views.py
Normal file
@@ -0,0 +1,110 @@
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from django.http import HttpResponse, JsonResponse
|
||||
|
||||
import swisseph as swe
|
||||
|
||||
from .calc import (
|
||||
DEFAULT_HOUSE_SYSTEM,
|
||||
get_element_counts,
|
||||
get_julian_day,
|
||||
get_planet_positions,
|
||||
set_ephe_path,
|
||||
)
|
||||
from .models import EphemerisSnapshot
|
||||
|
||||
|
||||
def chart(request):
|
||||
dt_str = request.GET.get('dt')
|
||||
lat_str = request.GET.get('lat')
|
||||
lon_str = request.GET.get('lon')
|
||||
|
||||
if not dt_str or lat_str is None or lon_str is None:
|
||||
return HttpResponse(status=400)
|
||||
|
||||
try:
|
||||
dt = datetime.fromisoformat(dt_str.replace('Z', '+00:00'))
|
||||
except ValueError:
|
||||
return HttpResponse(status=400)
|
||||
|
||||
try:
|
||||
lat = float(lat_str)
|
||||
lon = float(lon_str)
|
||||
except ValueError:
|
||||
return HttpResponse(status=400)
|
||||
|
||||
if not (-90 <= lat <= 90):
|
||||
return HttpResponse(status=400)
|
||||
|
||||
house_system_param = request.GET.get('house_system')
|
||||
if house_system_param is not None:
|
||||
if not (hasattr(request, 'user') and request.user.is_authenticated
|
||||
and request.user.is_superuser):
|
||||
return HttpResponse(status=403)
|
||||
house_system = house_system_param
|
||||
else:
|
||||
house_system = DEFAULT_HOUSE_SYSTEM
|
||||
|
||||
set_ephe_path()
|
||||
|
||||
jd = get_julian_day(dt)
|
||||
planets = get_planet_positions(jd)
|
||||
|
||||
cusps, ascmc = swe.houses(jd, lat, lon, house_system.encode())
|
||||
houses = {
|
||||
'cusps': list(cusps),
|
||||
'asc': ascmc[0],
|
||||
'mc': ascmc[1],
|
||||
}
|
||||
|
||||
return JsonResponse({
|
||||
'planets': planets,
|
||||
'houses': houses,
|
||||
'elements': get_element_counts(planets),
|
||||
'house_system': house_system,
|
||||
})
|
||||
|
||||
|
||||
def charts_list(request):
|
||||
date_from_str = request.GET.get('date_from')
|
||||
date_to_str = request.GET.get('date_to')
|
||||
|
||||
if not date_from_str or not date_to_str:
|
||||
return HttpResponse(status=400)
|
||||
|
||||
try:
|
||||
date_from = datetime.strptime(date_from_str, '%Y-%m-%d').replace(
|
||||
tzinfo=timezone.utc)
|
||||
date_to = datetime.strptime(date_to_str, '%Y-%m-%d').replace(
|
||||
hour=23, minute=59, second=59, tzinfo=timezone.utc)
|
||||
except ValueError:
|
||||
return HttpResponse(status=400)
|
||||
|
||||
if date_to < date_from:
|
||||
return HttpResponse(status=400)
|
||||
|
||||
qs = EphemerisSnapshot.objects.filter(dt__gte=date_from, dt__lte=date_to)
|
||||
|
||||
element_fields = {
|
||||
'fire_min': 'fire', 'water_min': 'water',
|
||||
'earth_min': 'earth', 'air_min': 'air',
|
||||
'time_min': 'time_el', 'space_min': 'space_el',
|
||||
}
|
||||
for param, field in element_fields.items():
|
||||
value = request.GET.get(param)
|
||||
if value is not None:
|
||||
try:
|
||||
qs = qs.filter(**{f'{field}__gte': int(value)})
|
||||
except ValueError:
|
||||
return HttpResponse(status=400)
|
||||
|
||||
results = [
|
||||
{
|
||||
'dt': snap.dt.isoformat(),
|
||||
'elements': snap.elements_dict(),
|
||||
'planets': snap.chart_data.get('planets', {}),
|
||||
}
|
||||
for snap in qs
|
||||
]
|
||||
|
||||
return JsonResponse({'results': results, 'count': len(results)})
|
||||
Reference in New Issue
Block a user