import datetime
import json
import time
from concurrent.futures import ThreadPoolExecutor
from flask import Flask, Response, render_template_string
import requests
app = Flask(__name__)
TOKEN_URL = "https://auth.opensky-network.org/auth/realms/opensky-network/protocol/openid-connect/token"
CLIENT_ID = "vaikero-api-client"
CLIENT_SECRET = "0X5576uw9airrXsR4ORMkujDoT3J5N87"
TOKEN_REFRESH_MARGIN = 30
PARAMS = {"lamin": 60.23, "lamax": 62.03, "lomin": 19.58, "lomax": 23.42}
API_URL = "https://opensky-network.org/api/states/all"
METADATA_URL = "https://opensky-network.org/api/metadata/aircraft/icao/"
# Globaali välimuisti ilmatilan tilanteesta ja koneiden malleista (vähentää API-kuormaa)
ilmatila_data = {"koneet": [], "paivitetty": "Ei vielä päivitetty", "status": "Käynnistetään..."}
viimeisimmät_icao_tunnukset = set()
malli_vallimuisti = {}
from datetime import datetime, timedelta
class TokenManager:
def __init__(self):
self.token = None
self.expires_at = None
def get_token(self):
"""Palauttaa voimassa olevan tokenin, hakee tarvittaessa uuden automaattisesti."""
if self.token and self.expires_at and datetime.now() < self.expires_at:
return self.token
return self._refresh()
def _refresh(self):
"""Hakee uuden access tokenin OpenSkyn palvelimelta."""
r = requests.post(
TOKEN_URL,
data={
"grant_type": "client_credentials",
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
},
timeout=10
)
r.raise_for_status()
data = r.json()
self.token = data["access_token"]
expires_in = data.get("expires_in", 1800)
self.expires_at = datetime.now() + timedelta(seconds=expires_in - TOKEN_REFRESH_MARGIN)
return self.token
def headers(self):
"""Palauttaa valmiit otsikot Bearer-tokenilla."""
return {"Authorization": f"Bearer {self.get_token()}"}
# Luodaan yksi jaettu manageri sovelluksen käyttöön
tokens = TokenManager()
def hae_koneen_malli(icao24):
"""Hakee koneen mallitiedot OpenSkyn metadatasta. Välimuistittaa tuloksen."""
if icao24 in malli_vallimuisti:
return malli_vallimuisti[icao24]
try:
# Haetaan yksittäisen koneen tarkemmat tiedot
res = requests.get(f"{METADATA_URL}{icao24}", auth=(OPENSKY_USER, OPENSKY_PASS), timeout=3)
if res.status_code == 200:
m_data = res.json()
valmistaja = m_data.get("manufacturerName", "")
malli = m_data.get("model", "")
tyyppikoodi = m_data.get("typecode", "")
if valmistaja or malli:
tulos = f"{valmistaja} {malli} ({tyyppikoodi})".strip()
else:
tulos = "Tuntematon konetyyppi"
malli_vallimuisti[icao24] = tulos
return tulos
except Exception:
pass
return "Ei saatavilla"
def paivita_ilmatila():
"""Taustafunktio, joka hakee datan OpenSkyltä OAuth2-tunnistautuneena."""
global ilmatila_data, viimeisimmät_icao_tunnukset
print("Taustaseuranta käynnistetty OAuth2-tunnistautumisella.")
while True:
nykyhetki = datetime.datetime.now().strftime("%H:%M:%S")
# 1. Haetaan tuore token
token = hae_access_token()
if not token:
ilmatila_data["status"] = "Tunnistautumisvirhe: Tokenia ei saatu."
print(f"[{nykyhetki}] ❌ Ei saatu tokenia. Odotetaan hetki...")
time.sleep(15)
continue
# 2. Luodaan otsikko haulle
headers = {
"Authorization": f"Bearer {token}"
}
try:
# Pääkysely ilmatilan tilasta virallisilla tunnuksilla
response = requests.get(
API_URL,
params=PARAMS,
headers=headers,
timeout=10
)
if response.status_code == 200:
data = response.json()
states = data.get("states") or []
nykyiset_koneet = []
nykyiset_icaot = set()
uusi_kone_havaittu = False
for flight in states:
icao24 = flight[0]
callsign = flight[1].strip() if flight[1] else "EI TUNNUSTA"
country = flight[2]
altitude = flight[7]
velocity = flight[9]
alt_ft = int(altitude * 3.28084) if altitude else None
speed_kmh = int(velocity * 3.6) if velocity else None
nykyiset_icaot.add(icao24)
if icao24 not in viimeisimmät_icao_tunnukset:
uusi_kone_havaittu = True
koneen_malli = "Ladataan..."
if icao24 in malli_vallimuisti:
koneen_malli = malli_vallimuisti[icao24]
else:
# Käytetään samaa tokenia myös metadatan hakuun, jos funktiosi sitä vaatii
koneen_malli = hae_koneen_malli(icao24)
time.sleep(1)
fr24_linkki = f"https://www.flightradar24.com/data/flights/{callsign.lower()}" if callsign != "EI TUNNUSTA" else "#"
adsb_linkki = f"https://globe.adsbexchange.com/?icao={icao24}"
nykyiset_koneet.append(
{
"icao": icao24,
"callsign": callsign,
"malli": koneen_malli,
"country": country,
"altitude": f"{alt_ft} ft (~{int(altitude)} m)" if alt_ft else "Ei saatavilla",
"speed": f"{speed_kmh} km/h" if speed_kmh else "Ei saatavilla",
"fr24": fr24_linkki,
"adsb": adsb_linkki
}
)
viimeisimmät_icao_tunnukset = nykyiset_icaot
ilmatila_data = {
"koneet": nykyiset_koneet,
"paivitetty": nykyhetki,
"status": "Yhteys kunnossa (OAuth2)",
"hälytys": uusi_kone_havaittu,
}
print(f"[{nykyhetki}] OpenSky päivitetty. Alueella {len(nykyiset_koneet)} konetta.")
elif response.status_code == 429:
ilmatila_data["status"] = "OpenSky virhe: 429 (Liikaa kyselyitä). Odotetaan..."
print(f"[{nykyhetki}] ⚠️ Palvelin palautti 429 tunnuksista huolimatta. Otetaan aikalisä.")
time.sleep(30)
else:
ilmatila_data["status"] = f"OpenSky virhe: {response.status_code}."
print(f"[{nykyhetki}] Virhe rajapinnassa: {response.status_code}")
except Exception as e:
ilmatila_data["status"] = f"Yhteysvirhe: {e}"
# Pidetään sovittu 30 sekunnin väli
time.sleep(30)
HTML_SIVU = """
Rauman Ilmatilavahti (100km)
✈️ Rauman Ilmatilavahti (100 km)
⚠️ Napsauta tästä aktivoidaksesi äänihälytykset
Odotetaan ensimmäistä päivitystä...
"""
@app.route("/")
def index():
return render_template_string(HTML_SIVU)
@app.route("/stream")
def stream():
def event_stream():
viimeisin_paivitys = ""
while True:
if ilmatila_data["paivitetty"] != viimeisin_paivitys:
yield f"data: {json.dumps(ilmatila_data)}\n\n"
viimeisin_paivitys = ilmatila_data["paivitetty"]
ilmatila_data["hälytys"] = False
time.sleep(1)
return Response(event_stream(), mimetype="text/event-stream")
import ssl
from concurrent.futures import ThreadPoolExecutor
if __name__ == "__main__":
executor = ThreadPoolExecutor(max_workers=1)
executor.submit(paivita_ilmatila)
cert_path = "/etc/letsencrypt/live/vibecoding.vaikkinen.info/fullchain.pem"
key_path = "/etc/letsencrypt/live/vibecoding.vaikkinen.info/privkey.pem"
print("Luodaan tiukka SSL-konteksti...")
# Luodaan explicit-konteksti palvelinkäyttöön
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(certfile=cert_path, keyfile=key_path)
print("Pakotetaan HTTPS-tila lokeissa näkyvälle domainille...")
app.run(debug=False, host="0.0.0.0", port=5000, ssl_context=context)