pytekstitv/pytekstitv.py

450 lines
13 KiB
Python
Executable file

#!/usr/bin/env python3
import argparse
import configparser
import dataclasses
import enum
import os
import select
import socket
import sys
import termios
import threading
import time
import urllib.request
import xml.etree.ElementTree
@dataclasses.dataclass
class Metadata:
lähde: str
edellinen: int | None
seuraava: int | None
alasivuja: int
def asetustiedosto():
asetuskansio = os.path.expanduser('~/.config')
if 'XDG_CONFIG_HOME' in os.environ:
asetuskansio = os.environ['XDG_CONFIG_HOME']
return os.path.join(asetuskansio, 'pytekstitv.conf')
def apikutsu(apiavain, sivu):
url = f'https://external.api.yle.fi/v1/teletext/pages/{sivu}.xml?{apiavain}'
with urllib.request.urlopen(url) as f:
return xml.etree.ElementTree.parse(f)
class Signaali:
def __init__(self):
self.kirjoituspää, self.lukupää = socket.socketpair()
def lähetä(self):
self.kirjoituspää.send(b'.')
def kuittaa(self):
self.lukupää.recv(1)
def odota(self):
poll = select.poll()
poll.register(self.lukupää, select.POLLIN)
poll.poll(-1) # Odota kunnes dataa on luettavaksi
self.kuittaa()
def fileno(self):
return self.lukupää.fileno()
class Jono:
def __init__(self):
self.jono_lukko = threading.Lock()
self.jono = []
self.signaali = Signaali()
def lähetä(self, viesti):
with self.jono_lukko:
self.jono.append(viesti)
self.signaali.lähetä()
def vastaanota(self):
self.signaali.odota()
with self.jono_lukko:
return self.jono.pop(0)
def tyhjä(self):
poll = select.poll()
poll.register(self.signaali, select.POLLIN)
return len(poll.poll(0)) == 0
class sivustatus(enum.Enum):
puuttuu, ei_sivua, epäonnistui, ok = range(4)
class LataajaSäie(threading.Thread):
def __init__(self, apiavain, välimuisti):
threading.Thread.__init__(self)
self.apiavain = apiavain
self.välimuisti = välimuisti
self.liian_tiuhaan = 0
def lataa(self, sivu):
if self.välimuisti.status(sivu) == sivustatus.ok:
return
try:
puu = apikutsu(self.apiavain, sivu)
self.välimuisti.lisää(sivu, prosessoi_sivu(puu))
self.liian_tiuhaan = 0
except urllib.error.HTTPError as virhe:
if virhe.code == 401:
self.liian_tiuhaan += 1
time.sleep(2 ** self.liian_tiuhaan)
if self.liian_tiuhaan > 4:
self.välimuisti.lisää(sivu, sivustatus.epäonnistui)
self.liian_tiuhaan = 0
elif virhe.code == 404:
self.välimuisti.lisää(sivu, sivustatus.ei_sivua)
self.liian_tiuhaan = 0
def run(self):
taustalla = []
while True:
while self.välimuisti.pyynnöt.tyhjä() and len(taustalla) > 0:
self.lataa(taustalla.pop())
match self.välimuisti.pyynnöt.vastaanota():
case None:
break
case [sivu, signaali]:
self.lataa(sivu)
signaali.lähetä()
taustalla = []
case sivu:
taustalla.append(sivu)
class Välimuisti:
def __init__(self, apiavain):
self.sivut_lukko = threading.Lock()
self.sivut = {}
self.pyynnöt = Jono()
LataajaSäie(apiavain, self).start()
def lisää(self, sivu, data):
with self.sivut_lukko:
self.sivut[sivu] = data
def status(self, sivu):
with self.sivut_lukko:
if sivu not in self.sivut:
return sivustatus.puuttuu
elif isinstance(self.sivut[sivu], sivustatus):
return self.sivut[sivu]
else:
return sivustatus.ok
def lataa(self, sivu, signaali):
self.pyynnöt.lähetä((sivu, signaali))
def linkki(self, sivu):
if sivu is not None and self.status(sivu) != sivustatus.ok:
self.pyynnöt.lähetä(sivu)
def hae(self, sivu):
metadata, alasivut = self.sivut[sivu]
return metadata, alasivut
# with-lauseketta varten
def __enter__(self):
return self
def __exit__(self, _1, _2, _3):
self.pyynnöt.lähetä(None)
# Heitä poikkeus uudestaan
return False
värit = {
'black': 30,
'red': 31,
'green': 32,
'yellow': 93,
'blue': 34,
'magenta': 95,
'cyan': 96,
'white': 97,
}
def grafiikkamerkki(arvo):
assert 0x20 <= arvo <= 0x7f
if arvo == 0x20: return '\u00a0' # No-Break Space
elif arvo < 0x35: return chr(0x1fb00 - 1 - 0x20 + arvo)
elif arvo == 0x35: return '\u258c' # Left Half Block
elif arvo <= 0x3f: return chr(0x1fb00 - 2 - 0x20 + arvo)
elif arvo < 0x60: return chr(arvo)
elif arvo < 0x6a: return chr(0x1fb00 - 2 - 0x40 + arvo)
elif arvo == 0x6a: return '\u2590' # Right Half Block
elif arvo < 0x7f: return chr(0x1fb00 - 3 - 0x40 + arvo)
elif arvo == 0x7f: return '\u2588' # Full Block
def ohjauskoodit(jakso):
grafiikka = False
väri = värit.get(jakso.get('fg'))
if väri == None:
assert jakso.get('fg')[0] == 'g'
väri = värit[jakso.get('fg')[1:]]
grafiikka = True
tausta = värit[jakso.get('bg')] + 10
teksti = jakso.text
if grafiikka:
teksti = ''.join(grafiikkamerkki(ord(merkki)) for merkki in teksti)
return f'\x1b[{väri};{tausta}m{teksti}'
def prosessoi_sivu(puu):
sivu = puu.getroot().find('page')
metadata = Metadata(
lähde = puu.getroot().get('network'),
edellinen = int(sivu.get('prevpg')) if sivu.get('prevpg') is not None else None,
seuraava = int(sivu.get('nextpg')) if sivu.get('nextpg') is not None else None,
alasivuja = int(sivu.get('subpagecount')),
)
alasivut = []
for alasivunumero in range(1, metadata.alasivuja + 1):
alasivu = []
for rivinumero in range(1, 24 + 1):
rivi = sivu.find(f'subpage[@number="{alasivunumero}"]/content[@type="structured"]/line[@number="{rivinumero}"]')
jaksot = [ohjauskoodit(jakso) for jakso in rivi]
alasivu.append(''.join(jaksot))
alasivut.append(alasivu)
return metadata, alasivut
def luo_käyttöliittymä(metadata, sivunumero, alasivunumero, sivunumero_syöte):
rivit = []
if len(sivunumero_syöte) == 0:
rivit.append(f'Sivu {sivunumero}')
else:
rivit.append(f'Sivu {sivunumero_syöte}{"_"*(3 - len(sivunumero_syöte))}')
if metadata.alasivuja != 1:
rivit.append(f'Alasivu {alasivunumero}/{metadata.alasivuja}')
else:
rivit.append('')
edellinen = metadata.edellinen if metadata.edellinen else ' '
seuraava = metadata.seuraava if metadata.seuraava else ' '
rivit.append(f'<<< {edellinen} {seuraava} >>>')
rivit.append('')
rivit.append(f'Sisällön lähde: {metadata.lähde}')
return rivit
def näytä_alasivu(metadata, alasivut, sivunumero, alasivunumero, sivunumero_syöte):
print('\x1b[1;1H\x1b[2J', end='') # Kursori yläkulmaan, tyhjennä näyttö
alasivu = alasivut[alasivunumero-1]
käyttöliittymä = luo_käyttöliittymä(metadata, sivunumero, alasivunumero, sivunumero_syöte)
korkeus = max(len(alasivu), len(käyttöliittymä))
for rivi in range(korkeus):
sisältö_rivi = alasivu[rivi] if rivi < len(alasivu) else ' '*40
käyttöliittymä_rivi = käyttöliittymä[rivi] if rivi < len(käyttöliittymä) else ''
print(sisältö_rivi, end='')
print('\x1b[0m ', end='')
print(käyttöliittymä_rivi, end='')
print('' if rivi == 23 else '\n', end='', flush = True)
class Sekvensoija:
def __init__(self):
self.puskuri = []
def syötä(self, merkki):
self.puskuri.append(merkki)
if self.puskuri[0] != b'\x1b':
# Yksittäinen merkki
return self.tyhjennä()
if len(self.puskuri) > 1 and self.puskuri[1] != b'[':
# Alt+näppäin, <esc><merkki>
return self.tyhjennä()
sarja1_alku = 2
# Oletus, että näppäinkoodi ei voi olla yli kolminumeroinen
sarja1_loppu = self.numerosarja(sarja1_alku, 3)
if len(self.puskuri) > sarja1_loppu and self.puskuri[sarja1_loppu] != b';':
# Erikoisnäppäin, <esc>[<muokkausnäppäimet><merkki>
return self.tyhjennä()
sarja2_alku = sarja1_loppu + 1
# Muokkausnäppäinkentän arvo voi olla maksimissaan 16
sarja2_loppu = self.numerosarja(sarja2_alku, 2)
if len(self.puskuri) > sarja2_loppu:
# Erikoisnäppäin, <esc>[<näppäinkoodi>;<muokkausnäppäimet>~
# Vaihtoehtoisesti virheellinen sarja, <esc>[<numeroita>;<numeroita><ei-tilde>
return self.tyhjennä()
return b''
def numerosarja(self, alku, maksimipituus):
pituus = min(maksimipituus, len(self.puskuri) - alku)
while pituus > 0:
if all(ord('0') <= tavu <= ord('9') for (tavu,) in self.puskuri[alku:alku + pituus]):
return alku + pituus
pituus -= 1
return alku
def tyhjennä(self):
syöte = b''.join(self.puskuri)
self.puskuri = []
return syöte
class päivitys(enum.Enum):
ei, piirrä, lataa = range(3)
def pääsilmukka(välimuisti, sivu_ladattu, sivunumero, alasivunumero = 1):
käynnissä = True
päivitä = päivitys.lataa
sekvensoija = Sekvensoija()
edellinen_sivunumero = None
sivunumero_syöte = ''
while käynnissä:
if päivitä == päivitys.lataa:
status = välimuisti.status(sivunumero)
if status == sivustatus.ok:
metadata, alasivut = välimuisti.hae(sivunumero)
välimuisti.linkki(metadata.edellinen)
välimuisti.linkki(metadata.seuraava)
päivitä = päivitys.piirrä
elif status == sivustatus.puuttuu:
välimuisti.lataa(sivunumero, sivu_ladattu)
päivitä = päivitys.ei
else:
sivunumero = edellinen_sivunumero
päivitä = päivitys.piirrä
if päivitä == päivitys.piirrä:
if alasivunumero < 0:
alasivunumero = metadata.alasivuja + 1 + alasivunumero
näytä_alasivu(metadata, alasivut, sivunumero, alasivunumero, sivunumero_syöte)
päivitä = päivitys.ei
poll = select.poll()
poll.register(sys.stdin.fileno(), select.POLLIN)
poll.register(sivu_ladattu, select.POLLIN)
for fd, tapahtuma in poll.poll():
assert tapahtuma == select.POLLIN
if fd == sivu_ladattu.fileno():
sivu_ladattu.kuittaa()
päivitä = päivitys.lataa
continue
assert fd == sys.stdin.fileno()
syöte = sekvensoija.syötä(sys.stdin.buffer.raw.read(1))
if syöte == b'': continue
if syöte in b'0123456789':
sivunumero_syöte += syöte.decode()
päivitä = päivitys.piirrä
if len(sivunumero_syöte) == 3:
edellinen_sivunumero = sivunumero
sivunumero = int(sivunumero_syöte)
alasivunumero = 1
sivunumero_syöte = ''
päivitä = päivitys.lataa
elif syöte == b'\x7f' and len(sivunumero_syöte) > 0:
sivunumero_syöte = sivunumero_syöte[:-1]
päivitä = päivitys.piirrä
elif len(sivunumero_syöte) > 0:
sivunumero_syöte = ''
päivitä = päivitys.piirrä
elif syöte == b'q':
käynnissä = False
elif syöte in (b'+', b'\x1b[B') and alasivunumero < metadata.alasivuja:
alasivunumero += 1
päivitä = päivitys.piirrä
elif syöte in (b'-', b'\x1b[A') and alasivunumero > 1:
alasivunumero -= 1
päivitä = päivitys.piirrä
elif syöte in (b'n', b'\x1b[C') and metadata.seuraava is not None:
edellinen_sivunumero = sivunumero
sivunumero = metadata.seuraava
alasivunumero = 1
päivitä = päivitys.lataa
elif syöte in (b'p', b'\x1b[D') and metadata.edellinen is not None:
edellinen_sivunumero = sivunumero
sivunumero = metadata.edellinen
alasivunumero = 1
päivitä = päivitys.lataa
elif syöte in (b' ', b'f', b'\x1b[6~'):
if alasivunumero < metadata.alasivuja:
alasivunumero += 1
päivitä = päivitys.piirrä
elif metadata.seuraava is not None:
edellinen_sivunumero = sivunumero
sivunumero = metadata.seuraava
alasivunumero = 1
päivitä = päivitys.lataa
elif syöte in (b'b', b'\x1b[5~'):
if alasivunumero > 1:
alasivunumero -= 1
päivitä = päivitys.piirrä
elif metadata.edellinen is not None:
edellinen_sivunumero = sivunumero
sivunumero = metadata.edellinen
alasivunumero = -1
päivitä = päivitys.lataa
def pää():
parseri = argparse.ArgumentParser()
parseri.add_argument('sivu', nargs='?')
sivunumero = parseri.parse_args().sivu
asetukset = configparser.ConfigParser()
asetukset.read(asetustiedosto())
if 'yle' not in asetukset or 'apiavain' not in asetukset['yle']:
print('Virhe: API-avain puuttuu.\n', file=sys.stderr)
print('1. Hanki API-avain osoitteesta https://tunnus.yle.fi/api-avaimet', file=sys.stderr)
print(f'2. Lisää tiedostoon {asetustiedosto()} osio:', file=sys.stderr)
print('[yle]', file=sys.stderr)
print('apiavain=app_id=…&app_key=…', file=sys.stderr)
sys.exit(1)
if sivunumero == None:
sivunumero = 100
else:
try:
sivunumero = int(sivunumero)
except ValueError:
print('Virhe: virheellinen sivunumero', file=sys.stderr)
sys.exit(1)
if sivunumero < 100 or 899 < sivunumero:
print('Virhe: sivunumeron tulee olla väliltä 100-899', file=sys.stderr)
sys.exit(1)
with Välimuisti(asetukset['yle']['apiavain']) as välimuisti:
sivu_ladattu = Signaali()
välimuisti.lataa(sivunumero, sivu_ladattu)
sivu_ladattu.odota()
if välimuisti.status(sivunumero) == sivustatus.ei_sivua:
print(f'Virhe: Sivua {sivunumero} ei löydy', file=sys.stderr)
sys.exit(1)
try:
print('\x1b[?1049h', end='') # Vaihtoehtoinen näyttöpuskuri
print('\x1b[?25l', end='') # Piilota kursori
try:
pääte = sys.stdout.fileno()
vanhat_attribuutit = termios.tcgetattr(pääte)
iflag, oflag, cflag, lflag, ispeed, ospeed, cc = vanhat_attribuutit
lflag &= ~termios.ECHO
lflag &= ~termios.ICANON
uudet_attribuutit = [iflag, oflag, cflag, lflag, ispeed, ospeed, cc]
termios.tcsetattr(pääte, termios.TCSADRAIN, uudet_attribuutit)
pääsilmukka(välimuisti, sivu_ladattu, sivunumero)
except KeyboardInterrupt:
pass
finally:
termios.tcsetattr(pääte, termios.TCSADRAIN, vanhat_attribuutit)
finally:
print('\x1b[?25h', end='') # Palauta kursori näkyväksi
print('\x1b[?1049l', end='', flush=True) # Palauta normaali näyttöpuskuri
if __name__ == '__main__':
pää()