Les fichiers

1. "w" vide le fichier des l'ouverture ; "a" place le curseur a la fin et conserve le contenu. Les deux creent le fichier s'il n'existe pas.
2. 4 : trois caracteres plus le "\n". C'est la cause d'innombrables comparaisons qui echouent (ligne == "abc" est faux).
3. write exige une chaine : TypeError: write() argument must be str. Il faut ecrire f.write(str(42)) ou print(42, file=f).
4. read() tente de charger 10 Go en memoire : MemoryError sur la plupart des machines. Le parcours ligne par ligne n'utilise que la memoire d'une ligne.
5. Le fichier reste ouvert jusqu'a ce que le ramasse-miettes le ferme, a un moment imprevisible. En ecriture, les donnees encore dans le tampon peuvent etre perdues si le programme s'interrompt. Sur un grand nombre de fichiers, on epuise les descripteurs (OSError: too many open files).

from collections import deque

def compte_lignes(nom):
    """Nombre de lignes. Theta(N) temps, Theta(1) memoire."""
    n = 0
    with open(nom, encoding="utf-8") as f:
        for _ in f:
            n += 1
    return n

def compte_mots(nom):
    n = 0
    with open(nom, encoding="utf-8") as f:
        for ligne in f:
            n += len(ligne.split())
    return n

def compte_caracteres(nom):
    n = 0
    with open(nom, encoding="utf-8") as f:
        for ligne in f:
            n += len(ligne)
    return n

def lignes_contenant(nom, motif):
    """Couples (numero de ligne, ligne) contenant le motif."""
    res = []
    with open(nom, encoding="utf-8") as f:
        for numero, ligne in enumerate(f, start=1):
            if motif in ligne:
                res.append((numero, ligne.rstrip("\n")))
    return res

def tete(nom, k=10):
    """k premieres lignes, sans lire le reste du fichier."""
    res = []
    with open(nom, encoding="utf-8") as f:
        for ligne in f:
            res.append(ligne.rstrip("\n"))
            if len(res) == k:
                break                     # on s'arrete des que possible
    return res

def queue(nom, k=10):
    """k dernieres lignes, en memoire Theta(k)."""
    with open(nom, encoding="utf-8") as f:
        return [l.rstrip("\n") for l in deque(f, maxlen=k)]

# test
with open("_t.txt", "w", encoding="utf-8") as f:
    for i in range(1, 101):
        f.write(f"ligne {i} du fichier\n")

assert compte_lignes("_t.txt") == 100
assert compte_mots("_t.txt") == 400
assert tete("_t.txt", 2) == ["ligne 1 du fichier", "ligne 2 du fichier"]
assert queue("_t.txt", 2) == ["ligne 99 du fichier", "ligne 100 du fichier"]
# "ligne 5" apparait dans "ligne 5" et dans "ligne 50" a "ligne 59" : 1 + 10 = 11
assert len(lignes_contenant("_t.txt", "ligne 5")) == 11
assert len(lignes_contenant("_t.txt", "ligne 5 ")) == 1     # avec l'espace : une seule
import os; os.remove("_t.txt")

4. deque(f, maxlen=k) est l'astuce : la deque ne conserve jamais plus de \(k\) elements, les plus anciens sortant automatiquement. On parcourt donc le fichier une seule fois, en memoire \(\Theta(k)\) au lieu de \(\Theta(N)\).

Une vraie commande tail fait mieux encore : elle se positionne pres de la fin du fichier avec f.seek(-taille, 2) et ne lit que les derniers octets, en \(\Theta(k)\) temps et non \(\Theta(N)\).

import os

CONTENU = """date,ville,temperature,pluie
2024-01-01,Rabat,14.2,0
2024-01-01,Fes,9.8,3.5
2024-01-02,Rabat,15.1,0
2024-01-02,Fes,11.2,1.2
2024-01-03,Rabat,abc,0
2024-01-03,Fes,10.5
"""

with open("meteo.csv", "w", encoding="utf-8") as f:
    f.write(CONTENU)

def charge(nom):
    """Liste de dictionnaires typees, lignes malformees ignorees et signalees."""
    donnees, rejets = [], []
    with open(nom, encoding="utf-8") as f:
        entetes = f.readline().rstrip("\n").split(",")
        for numero, ligne in enumerate(f, start=2):
            ligne = ligne.rstrip("\n")
            if not ligne:
                continue
            champs = ligne.split(",")
            if len(champs) != len(entetes):
                rejets.append((numero, "nombre de champs incorrect"))
                continue
            e = dict(zip(entetes, champs))
            try:
                e["temperature"] = float(e["temperature"])
                e["pluie"] = float(e["pluie"])
            except ValueError as err:
                rejets.append((numero, f"conversion impossible : {err}"))
                continue
            donnees.append(e)
    return donnees, rejets

def resume(donnees):
    """Temperature moyenne et cumul de pluie par ville."""
    par_ville = {}
    for e in donnees:
        v = par_ville.setdefault(e["ville"], {"temps": [], "pluie": 0.0})
        v["temps"].append(e["temperature"])
        v["pluie"] += e["pluie"]
    return {ville: {"temperature_moyenne": sum(v["temps"]) / len(v["temps"]),
                    "pluie_totale": v["pluie"],
                    "releves": len(v["temps"])}
            for ville, v in par_ville.items()}

def ecrit_resume(nom, r):
    with open(nom, "w", encoding="utf-8") as f:
        f.write("ville,temperature_moyenne,pluie_totale,releves\n")
        for ville, v in sorted(r.items()):
            f.write(f"{ville},{v['temperature_moyenne']:.2f},"
                    f"{v['pluie_totale']:.1f},{v['releves']}\n")

donnees, rejets = charge("meteo.csv")
for numero, raison in rejets:
    print(f"ligne {numero} rejetee : {raison}")
r = resume(donnees)
ecrit_resume("resume.csv", r)
print(open("resume.csv", encoding="utf-8").read())

os.remove("meteo.csv"); os.remove("resume.csv")

ligne 6 rejetee : conversion impossible : could not convert string to float: 'abc'
ligne 7 rejetee : nombre de champs incorrect
ville,temperature_moyenne,pluie_totale,releves
Fes,10.50,4.7,2
Rabat,14.65,0.0,2

La separation entre charge (lecture et validation), resume (calcul pur) et ecrit_resume (sortie) permet de tester chaque partie independamment : la fonction resume n'a pas besoin de fichier pour etre testee.

import os

def fusionne(f1, f2, sortie):
    """Concatene deux fichiers. Memoire Theta(1)."""
    with open(sortie, "w", encoding="utf-8") as s:
        for nom in (f1, f2):
            with open(nom, encoding="utf-8") as f:
                for ligne in f:
                    s.write(ligne)
                if not ligne.endswith("\n"):     # eviter de coller deux lignes
                    s.write("\n")

def filtre(entree, sortie, predicat):
    """Conserve les lignes verifiant predicat(ligne_sans_saut)."""
    n = 0
    with open(entree, encoding="utf-8") as e, open(sortie, "w", encoding="utf-8") as s:
        for ligne in e:
            if predicat(ligne.rstrip("\n")):
                s.write(ligne)
                n += 1
    return n

def numerote(entree, sortie):
    with open(entree, encoding="utf-8") as e, open(sortie, "w", encoding="utf-8") as s:
        for i, ligne in enumerate(e, start=1):
            s.write(f"{i:5} | {ligne}")

def dedoublonne(entree, sortie):
    """Supprime les doublons, ordre de premiere apparition conserve. Memoire Theta(k)."""
    vues = set()
    with open(entree, encoding="utf-8") as e, open(sortie, "w", encoding="utf-8") as s:
        for ligne in e:
            cle = ligne.rstrip("\n")
            if cle not in vues:
                vues.add(cle)
                s.write(ligne)

# tests
with open("_a.txt", "w", encoding="utf-8") as f:
    f.write("alpha\nbeta\nalpha\ngamma\n")
dedoublonne("_a.txt", "_b.txt")
assert open("_b.txt", encoding="utf-8").read() == "alpha\nbeta\ngamma\n"
assert filtre("_a.txt", "_c.txt", lambda l: l.startswith("a")) == 2
numerote("_a.txt", "_d.txt")
print(open("_d.txt", encoding="utf-8").read())
for n in ("_a.txt", "_b.txt", "_c.txt", "_d.txt"):
    os.remove(n)

Toutes ces fonctions travaillent en flux : elles fonctionnent sur des fichiers de taille arbitraire. Seule dedoublonne consomme de la memoire proportionnelle au nombre de lignes distinctes. Pour un fichier trop volumineux, on trierait d'abord le fichier (tri externe) puis on supprimerait les doublons consecutifs, en memoire constante.

import os

JOURNAL = """2024-03-01 08:12:45 INFO connexion utilisateur=amine
2024-03-01 08:13:02 ERROR echec authentification utilisateur=inconnu
2024-03-01 08:15:30 WARNING disque plein a 91%
2024-03-01 09:01:11 ERROR timeout base de donnees
2024-03-01 09:22:40 ERROR echec authentification utilisateur=sara
2024-03-01 10:00:00 INFO sauvegarde terminee utilisateur=amine
"""
with open("_j.log", "w", encoding="utf-8") as f:
    f.write(JOURNAL)

def analyse(nom):
    """Compte les lignes par niveau, extrait erreurs, heures et utilisateurs."""
    niveaux, erreurs, par_heure, utilisateurs = {}, [], {}, set()
    with open(nom, encoding="utf-8") as f:
        for ligne in f:
            champs = ligne.rstrip("\n").split()
            if len(champs) < 3:
                continue
            date, heure, niveau = champs[0], champs[1], champs[2]
            message = " ".join(champs[3:])

            niveaux[niveau] = niveaux.get(niveau, 0) + 1
            if niveau == "ERROR":
                erreurs.append((date, heure, message))
                h = heure[:2]
                par_heure[h] = par_heure.get(h, 0) + 1
            for c in champs:
                if c.startswith("utilisateur="):
                    utilisateurs.add(c.split("=", 1)[1])
    return niveaux, erreurs, par_heure, utilisateurs

niveaux, erreurs, par_heure, utilisateurs = analyse("_j.log")
print(niveaux)                    # {'INFO': 2, 'ERROR': 3, 'WARNING': 1}
print(len(erreurs))               # 3
print(max(par_heure, key=par_heure.get))       # '09'
print(sorted(utilisateurs))       # ['amine', 'inconnu', 'sara']
os.remove("_j.log")

Le decoupage par split() suppose que les trois premiers champs ne contiennent pas d'espace, ce qui est le cas ici. Le split("=", 1) avec la limite 1 est important : il preserve un eventuel = dans la valeur.

Un seul parcours du fichier alimente les quatre statistiques : sur un journal de plusieurs gigaoctets, relire le fichier quatre fois serait quatre fois plus lent.

import os
from string import ascii_lowercase as ALPHABET

MOTS = ["chat", "chien", "chien", "chine", "chaton", "char", "chat", "chats", "niche"]
with open("_dico.txt", "w", encoding="utf-8") as f:
    f.write("\n".join(MOTS) + "\n")

def charge_dico(nom):
    """Ensemble des mots du dictionnaire. Theta(N) une fois, puis Theta(1) par test."""
    with open(nom, encoding="utf-8") as f:
        return {l.strip().lower() for l in f if l.strip()}

def mots_inconnus(texte, dico):
    """Mots du texte absents du dictionnaire, sans doublon."""
    propre = "".join(c if c.isalpha() else " " for c in texte.lower())
    return sorted({m for m in propre.split() if m not in dico})

def variantes(mot):
    """Tous les mots a distance d'edition 1 de mot."""
    v = set()
    for i in range(len(mot) + 1):
        g, d = mot[:i], mot[i:]
        if d:
            v.add(g + d[1:])                        # suppression
            if len(d) > 1:
                v.add(g + d[1] + d[0] + d[2:])      # transposition
            for c in ALPHABET:
                v.add(g + c + d[1:])                # substitution
        for c in ALPHABET:
            v.add(g + c + d)                        # insertion
    v.discard(mot)
    return v

def suggestions(mot, dico):
    """Mots du dictionnaire a distance d'edition 1."""
    return sorted(variantes(mot.lower()) & dico)

dico = charge_dico("_dico.txt")
assert mots_inconnus("Le chat et le chein dorment.", dico) == ["chein", "dorment", "et", "le"]
print(suggestions("chein", dico))      # ['chien']
print(suggestions("cha", dico))        # ['char', 'chat']
os.remove("_dico.txt")

1. Un dictionnaire francais comporte environ \(3\cdot10^5\) mots. Avec une liste, chaque test mot in dico couterait \(\Theta(N)\) : verifier un texte de \(10^4\) mots demanderait \(3\cdot10^9\) comparaisons. Avec un ensemble, chaque test est en \(\Theta(1)\) : \(10^4\) operations. Facteur \(3\cdot10^5\).

4. Pour un mot de longueur \(n\) sur un alphabet de taille \(a = 26\) :

Soit environ 300 candidats pour un mot de 5 lettres : c'est negligeable, et l'intersection avec le dictionnaire est en \(\Theta(1)\) par candidat. C'est exactement l'approche du correcteur de Peter Norvig. Pour la distance 2, on genererait \(\approx (54n)^2 \approx 10^5\) candidats, encore praticable.

1. Principe. Le fichier de \(N\) lignes ne tient pas en memoire, mais un bloc de \(B\) lignes, oui. On procede en deux phases :

• Phase de tri : lire \(B\) lignes, les trier en memoire, les ecrire dans un fichier temporaire. Repeter : on obtient \(\lceil N/B\rceil\) fichiers tries.
• Phase de fusion : fusionner ces fichiers en gardant une seule ligne de chacun en memoire, en prenant a chaque etape le minimum (grace a un tas).

import os, heapq, random

def decoupe(entree, taille_bloc):
    """Decoupe le fichier en blocs tries. Renvoie la liste des fichiers temporaires."""
    temporaires, bloc, k = [], [], 0
    with open(entree, encoding="utf-8") as f:
        for ligne in f:
            bloc.append(ligne.rstrip("\n"))
            if len(bloc) == taille_bloc:
                temporaires.append(_ecrit_bloc(bloc, k)); k += 1
                bloc = []
    if bloc:
        temporaires.append(_ecrit_bloc(bloc, k))
    return temporaires

def _ecrit_bloc(bloc, k):
    bloc.sort()
    nom = f"_bloc{k}.tmp"
    with open(nom, "w", encoding="utf-8") as f:
        f.write("\n".join(bloc) + "\n")
    return nom

def fusionne_fichiers(fichiers, sortie):
    """Fusionne k fichiers tries. Memoire Theta(k)."""
    ouverts = [open(n, encoding="utf-8") for n in fichiers]
    tas = []
    for i, f in enumerate(ouverts):
        ligne = f.readline()
        if ligne:
            heapq.heappush(tas, (ligne.rstrip("\n"), i))
    with open(sortie, "w", encoding="utf-8") as s:
        while tas:
            valeur, i = heapq.heappop(tas)         # minimum global
            s.write(valeur + "\n")
            suivante = ouverts[i].readline()       # on reapprovisionne ce flux
            if suivante:
                heapq.heappush(tas, (suivante.rstrip("\n"), i))
    for f in ouverts:
        f.close()

def tri_externe(entree, sortie, taille_bloc=1000):
    temporaires = decoupe(entree, taille_bloc)
    fusionne_fichiers(temporaires, sortie)
    for n in temporaires:
        os.remove(n)

# test
valeurs = [f"{random.randint(0, 999999):06d}" for _ in range(5000)]
with open("_gros.txt", "w", encoding="utf-8") as f:
    f.write("\n".join(valeurs) + "\n")

tri_externe("_gros.txt", "_trie.txt", taille_bloc=300)
resultat = [l.rstrip("\n") for l in open("_trie.txt", encoding="utf-8")]
assert resultat == sorted(valeurs)
print("tri externe valide :", len(resultat), "lignes")
os.remove("_gros.txt"); os.remove("_trie.txt")

4. La memoire utilisee est \(\Theta(B)\) pendant la phase de tri (un bloc) et \(\Theta(k)\) pendant la fusion (une ligne par fichier, plus le tas), avec \(k = \lceil N/B\rceil\). Elle est donc independante de \(N\), ce qui est tout l'interet.

En temps : le tri des blocs coute \(\Theta\!\left(\frac{N}{B} \cdot B\log B\right) = \Theta(N\log B)\) ; la fusion traite \(N\) lignes avec une operation de tas en \(\Theta(\log k)\) chacune, soit \(\Theta(N\log k)\). Total : \(\Theta(N\log N)\) comme un tri interne, mais avec un nombre d'acces disque minimal — c'est ce dernier critere qui compte pour de gros volumes.

import heapq, os
from collections import Counter

def frequences(texte):
    return Counter(texte)

def arbre_huffman(freq):
    """Construit l'arbre. Les noeuds sont (poids, compteur, contenu)."""
    tas = [(p, i, c) for i, (c, p) in enumerate(sorted(freq.items()))]
    heapq.heapify(tas)
    compteur = len(tas)
    if len(tas) == 1:                      # cas degenere : un seul caractere
        p, _, c = tas[0]
        return (p, 0, (c, None))
    while len(tas) > 1:
        p1, _, g = heapq.heappop(tas)      # les deux plus petits poids
        p2, _, d = heapq.heappop(tas)
        heapq.heappush(tas, (p1 + p2, compteur, (g, d)))
        compteur += 1
    return tas[0]

def table_codes(arbre):
    """Dictionnaire caractere -> code binaire, par parcours de l'arbre."""
    codes = {}
    def parcours(noeud, prefixe):
        if isinstance(noeud, str):         # feuille
            codes[noeud] = prefixe or "0"
            return
        g, d = noeud
        parcours(g, prefixe + "0")
        if d is not None:
            parcours(d, prefixe + "1")
    parcours(arbre[2], "")
    return codes

def code(texte, codes):
    return "".join(codes[c] for c in texte)

def decode(bits, arbre):
    """Decodage : on descend dans l'arbre bit a bit."""
    racine = arbre[2]
    res, noeud = [], racine
    for b in bits:
        if isinstance(noeud, str):
            res.append(noeud); noeud = racine
        noeud = noeud[0] if b == "0" else noeud[1]
        if isinstance(noeud, str):
            res.append(noeud); noeud = racine
    return "".join(res)

texte = "abracadabra" * 30
f = frequences(texte)
a = arbre_huffman(f)
codes = table_codes(a)
bits = code(texte, codes)

print({c: codes[c] for c in sorted(codes)})
print(f"original  : {len(texte) * 8} bits (8 bits par caractere)")
print(f"compresse : {len(bits)} bits")
print(f"taux      : {100 * (1 - len(bits) / (8*len(texte))):.1f} % d'economie")
assert decode(bits, a) == texte

{'a': '0', 'b': '111', 'c': '1101', 'd': '1100', 'r': '10'}
original  : 2640 bits (8 bits par caractere)
compresse : 690 bits
taux      : 73.9 % d'economie

2. Algorithme. On place chaque caractere dans une file de priorite avec son effectif comme cle. Tant qu'il reste plus d'un element, on extrait les deux plus petits, on cree un noeud interne de poids la somme, et on le reinsere. Le dernier element est la racine. Complexite : \(\Theta(a\log a)\) pour \(a\) caracteres distincts.

4. Code prefixe. Les caracteres sont places aux feuilles de l'arbre, et le code d'un caractere est le chemin de la racine a sa feuille. Or aucun chemin racine-feuille n'est prefixe d'un autre chemin racine-feuille (un prefixe correspondrait a un noeud interne, qui ne porte aucun caractere). Le decodage est donc non ambigu et se fait sans separateur, simplement en descendant dans l'arbre.

Optimalite et entropie. Huffman produit un code prefixe de longueur moyenne minimale. Le theoreme de Shannon donne l'encadrement \[ H(X) \;\le\; \bar{\ell} \;<\; H(X) + 1, \qquad H(X) = -\sum_{c} p_c \log_2 p_c, \] ou \(H\) est l'entropie de la source et \(\bar\ell\) la longueur moyenne des codes. Aucun codage sans perte ne peut descendre en dessous de \(H(X)\) bits par symbole en moyenne.

from math import log2
n = len(texte)
H = -sum((v/n) * log2(v/n) for v in f.values())
print(f"entropie : {H:.3f} bits/symbole")
print(f"Huffman  : {len(bits)/n:.3f} bits/symbole")
# entropie : 2.040   Huffman : 2.091   -> on est bien dans [H, H+1[