pytekstitv/pytekstitv.py

#!/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ää()