Montréal : lire les transformations locales

from pathlib import Path
from collections import Counter
import hashlib
import json
import math
import sys

import pandas as pd
import matplotlib.pyplot as plt
from IPython.display import display, Markdown, Image

candidates = [Path.cwd().resolve(), *Path.cwd().resolve().parents]
STUDY_ROOT = next((p for p in candidates if (p / 'PUBLICATION_EVIDENCE.json').is_file()
                   and (p / 'prototype/assets/study.json').is_file()), None)
if STUDY_ROOT is None:
    raise FileNotFoundError('Exécuter le notebook depuis le dossier research-deliverables de l’étude.')
DELIVERABLES = STUDY_ROOT / 'research-deliverables'
ASSETS = STUDY_ROOT / 'prototype/assets'

pd.set_option('display.max_colwidth', 100)
pd.set_option('display.max_rows', 30)
plt.rcParams.update({'figure.dpi': 125, 'font.family': 'DejaVu Sans',
                     'axes.spines.top': False, 'axes.spines.right': False})
print('Étude locale chargée. Les entrées sont lues sans modification.')

def read_json(path):
    return json.loads(path.read_text(encoding='utf-8'))

def sha256(path):
    digest = hashlib.sha256()
    with path.open('rb') as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b''):
            digest.update(chunk)
    return digest.hexdigest()

study = read_json(ASSETS / 'study.json')
evidence = read_json(STUDY_ROOT / 'PUBLICATION_EVIDENCE.json')
manifest = read_json(STUDY_ROOT / 'PUBLICATION_MANIFEST.json')
validation = read_json(STUDY_ROOT / 'PUBLICATION_VALIDATION.json')
assert validation['checks_passed'] == manifest['publication_checks']
assert validation['inputs_preserved'] is True
assert manifest['database_access'] == 'none'

source_checks = []
for relative, expected in manifest['input_sha256'].items():
    path = STUDY_ROOT / relative
    actual = sha256(path)
    assert actual == expected, f'Entrée modifiée : {path.name}'
    assert study['source_artifact_sha256'][path.name] == expected
    source_checks.append({'Artefact': path.name, 'SHA-256 vérifié': True})
assert sha256(STUDY_ROOT / 'PUBLICATION_EVIDENCE.json') == manifest['publication_files']['PUBLICATION_EVIDENCE.json']['sha256']
for name, recorded in manifest['figures'].items():
    assert sha256(ASSETS / 'figures' / name) == recorded['sha256'], name
for source in study['methodology_sources']['files']:
    assert sha256(ASSETS / 'methodology' / source['path']) == source['sha256']

cases = study['cases']
assert [c['tract_id'] for c in cases] == manifest['case_ids']
for c in cases:
    for key, expected in evidence[c['tract_id']].items():
        actual = c[key]
        if isinstance(expected, float):
            assert math.isclose(actual, expected, abs_tol=0.000001), (c['name'], key)
        else:
            assert actual == expected, (c['name'], key)
for key, expected in evidence['verdun'].items():
    assert study['verdun'][key] == expected, ('Verdun', key)
# Presentation-specific additions are independently checked against raw records.
raw_permits = pd.read_csv(STUDY_ROOT / 'data/deep_dive/permit_timeline.csv', dtype=str, keep_default_na=False).to_dict('records')
for record in study['verdun']['galt_records']:
    assert any(all(raw[k] == value for k, value in record.items()) for raw in raw_permits)
review = pd.read_csv(STUDY_ROOT / 'data/deep_dive/baseline_CO_description_review.csv', dtype=str, keep_default_na=False)
review = review[(review['tract_id'] == '4620305.00') & (review['housing_label_broad'] == 'True')]
classes = review['description_review_class'].value_counts()
assert len(review) == study['verdun']['housing_CO']
assert classes['principal_housing_construction_described'] == study['verdun']['principal_housing_descriptions']
assert classes['excavation_related_to_principal_building'] == study['verdun']['excavation_descriptions']
assert classes['other_work_no_new_principal_building_described'] == study['verdun']['other_descriptions']
assert study['relationship'] == evidence['relationships'][1]

print(f"Vérification : {len(source_checks)} empreintes analytiques, {len(manifest['figures'])} figures originales et 3 PDF de méthode.")
print(f"QA de publication existante : {validation['checks_passed']} contrôles et {validation['quantitative_claim_groups']} groupes de chiffres.")
display(pd.DataFrame(source_checks))

population = pd.DataFrame([{
    'Dossier': c['name'], 'Secteur 2021': c['tract_id'],
    'Résidents 2016': c['population_2016'], 'Résidents 2021': c['population_2021'],
    'Variation de population (%)': c['population_change_pct'],
} for c in cases]).set_index('Dossier')
for c in cases:
    calculated = 100 * (c['population_2021'] - c['population_2016']) / c['population_2016']
    assert math.isclose(calculated, c['population_change_pct'], abs_tol=1e-10)
display(population.style.format({'Variation de population (%)': '{:+.1f}'}))
display(Image(filename=str(ASSETS / 'figures/overview_map.png'), width=1000))

panel = pd.DataFrame([{
    'Dossier': c['name'], 'Locaux appariés n': c['panel_n'],
    'Occupés 2021': c['panel_occupied_2021'], 'Vacants 2021': c['panel_vacant_2021'],
    'Occupés 2025': c['panel_occupied_2025'], 'Vacants 2025': c['panel_vacant_2025'],
    'Vacant → occupé': c['vacant_to_occupied'], 'Occupé → vacant': c['occupied_to_vacant'],
    'Variation de part occupée (points)': c['panel_occupied_change_pp'],
} for c in cases]).set_index('Dossier')
for c in cases:
    n = c['panel_n']
    assert c['panel_occupied_2021'] + c['panel_vacant_2021'] == n
    assert c['panel_occupied_2025'] + c['panel_vacant_2025'] == n
    assert c['panel_occupied_2025'] == c['panel_occupied_2021'] + c['vacant_to_occupied'] - c['occupied_to_vacant']
    expected_pp = 100 * (c['panel_occupied_2025'] - c['panel_occupied_2021']) / n
    assert math.isclose(expected_pp, c['panel_occupied_change_pp'], abs_tol=1e-10)
    assert c['vacant_to_occupied'] <= c['panel_vacant_2021']
    assert c['occupied_to_vacant'] <= c['panel_occupied_2021']
display(panel.style.format({'Variation de part occupée (points)': '{:+.1f}'}))

fig, axes = plt.subplots(1, 3, figsize=(12, 4.4), sharex=True)
for ax, c in zip(axes, cases):
    n = c['panel_n']
    occupied = [100 * c[f'panel_occupied_{year}'] / n for year in [2021, 2025]]
    vacant = [100 * c[f'panel_vacant_{year}'] / n for year in [2021, 2025]]
    ax.barh([1, 0], occupied, color='#007166', height=.45, label='Occupé')
    ax.barh([1, 0], vacant, left=occupied, color='#ead6b0', edgecolor='#8c682c', height=.45, label='Vacant')
    ax.set_yticks([1, 0], ['2021', '2025'])
    ax.set_xlim(0, 100); ax.set_xticks([0, 50, 100]); ax.set_xlabel('Part des locaux appariés (%)')
    ax.set_title(f"{c['name']} · n={n}", loc='left', fontsize=11, fontweight='bold')
    ax.text(0, -.28, f"2021: {c['panel_occupied_2021']} occupés / {c['panel_vacant_2021']} {'vacant' if c['panel_vacant_2021'] == 1 else 'vacants'}\n"
                     f"2025: {c['panel_occupied_2025']} occupés / {c['panel_vacant_2025']} vacants", fontsize=9, va='top')
    ax.set_ylim(-.72, 1.5)
handles, labels = axes[0].get_legend_handles_labels()
fig.legend(handles, labels, loc='lower center', ncol=2, frameon=False)
fig.suptitle('Occupation commerciale · locaux appariés, 2021–2025', x=.06, ha='left', fontsize=14)
fig.subplots_adjust(left=.06, right=.98, top=.8, bottom=.2, wspace=.30)
plt.show(); plt.close(fig)

inventory = pd.DataFrame([{
    'Dossier': c['name'], 'Tous étages, 2021': c['all_floor_stock_2021'],
    'Tous étages, 2025': c['all_floor_stock_2025'],
    'Étages comparables, 2021': c['comparable_stock_2021'],
    'Étages comparables, 2025': c['comparable_stock_2025'],
} for c in cases]).set_index('Dossier')
display(inventory)

c = next(c for c in cases if c['key'] == 'lachine')
display(Image(filename=str(ASSETS / 'figures/lachine_case.png'), width=1000))
display(pd.DataFrame([{'Corridor': street, 'Locaux appariés changeant d’état': count}
                     for street, count in c['corridors'].items()]))

c = next(c for c in cases if c['key'] == 'ahuntsic')
display(Image(filename=str(ASSETS / 'figures/ahuntsic_case.png'), width=1000))
display(pd.DataFrame([{'Corridor': street, 'Locaux appariés changeant d’état': count}
                     for street, count in c['corridors'].items()]))

c = next(c for c in cases if c['key'] == 'plateau')
display(Image(filename=str(ASSETS / 'figures/plateau_case.png'), width=1000))
display(pd.DataFrame([{'Corridor': street, 'Locaux appariés changeant d’état': count}
                     for street, count in c['corridors'].items()]))
# Existing classification result; this is not a measure of sales or employment.
food = pd.DataFrame(study['plateau_food'])[['survey_edition', 'classified_occupied', 'share']]
food = food.rename(columns={'survey_edition': 'Édition de l’enquête', 'classified_occupied': 'Locaux occupés, alimentation et restauration', 'share': 'Part des locaux occupés classifiés'})
display(food.style.format({'Part des locaux occupés classifiés': '{:.1%}'}))

permit_rows = []
property_rows = []
for c in cases:
    for project in c['projects']:
        for record in project['permit_records']:
            permit_rows.append({'Dossier': c['name'], 'Adresse': project['address'],
                                'Date d’autorisation': record['issued_date'], 'Code': record['permit_type_code'],
                                'Champ brut des logements touchés': record['raw_dwellings_value'],
                                'Description originale des travaux': record['nature_travaux']})
    for record in c['assessments']:
        property_rows.append({'Dossier': c['name'], 'Caractéristiques immobilières': record['target'],
                              'Logements résidentiels consignés': record['current_recorded_dwelling_sum'],
                              'Années de construction inscrites': ', '.join(y for y in record['construction_years'] if y != '9999') or 'Non résolu',
                              'État modifié': record['stock_snapshot_last_modified'],
                              'Consulté': record['retrieved_date'], 'Confiance du rapprochement': record['join_confidence']})
display(pd.DataFrame(permit_rows).sort_values(['Dossier', 'Date d’autorisation']).reset_index(drop=True))
display(pd.DataFrame(property_rows))

history_rows = []
for c in cases:
    counts = Counter(r['transition'] for r in c['transition_premises'])
    assert counts['vacant→occupied'] == c['vacant_to_occupied']
    assert counts['occupied→vacant'] == c['occupied_to_vacant']
    dates = Counter()
    for record in c['transition_premises']:
        states = record['state_sequence'].split('→')
        assert len(states) == 5
        assert f'{states[0]}→{states[-1]}' == record['transition']
        changed_date = record.get('first_observed_occupied_after_baseline') or record.get('first_observed_vacant_after_baseline')
        dates[changed_date[:4]] += 1
        history_rows.append({'Dossier': c['name'], 'Locaux appariés n': c['panel_n'],
                             'Occupés / vacants 2021': f"{c['panel_occupied_2021']} / {c['panel_vacant_2021']}",
                             'Adresse': record['baseline_address'], 'Étage': record['baseline_floor'],
                             'Unité': record['baseline_suite'] or 'Non fourni',
                             **dict(zip(map(str, range(2021, 2026)), states)),
                             'Première observation du nouvel état': changed_date})
    assert dict(sorted(dates.items())) == c['first_observed_transition_year_counts']
histories = pd.DataFrame(history_rows)
display(histories.replace({'occupied': 'occupé', 'vacant': 'vacant', 'missing': 'non observé', 'unknown': 'inconnu'}))

relationship = study['relationship']
assert relationship['n'] == manifest['eligible_scatter_n'] == 267
print(f"Association déjà consignée : Spearman rho={relationship['spearman']:.3f}; n={relationship['n']}.")
display(Image(filename=str(ASSETS / 'figures/population_vs_commerce.png'), width=1000))
display(Image(filename=str(ASSETS / 'figures/three_case_comparison.png'), width=1000))

v = study['verdun']
assert v['housing_CO'] == v['principal_housing_descriptions'] + v['excavation_descriptions'] + v['other_descriptions']
permit_review = pd.DataFrame([
    ['Événements codés construction résidentielle', v['housing_CO']],
    ['Descriptions directes de construction principale', v['principal_housing_descriptions']],
    ['Excavations associées', v['excavation_descriptions']],
    ['Autres descriptions de travaux', v['other_descriptions']],
], columns=['Description, 2016–2020', 'Événements'])
display(permit_review)
display(pd.DataFrame(v['galt_records'])[['issued_date', 'nature_travaux', 'raw_dwellings_value']]
        .rename(columns={'issued_date': 'Date d’autorisation', 'nature_travaux': 'Description originale au 200 Galt', 'raw_dwellings_value': 'Champ des logements touchés'}))
verdun_inventory = pd.DataFrame([
    ['Tous les étages principaux observés', *v['all_floor_stock']],
    ['Deuxièmes étages exclus aux deux dates', *v['comparable_stock']],
], columns=['Périmètre de l’enquête', 'Locaux 2021', 'Locaux 2025'])
verdun_inventory['Variation de l’inventaire observé'] = verdun_inventory['Locaux 2025'] - verdun_inventory['Locaux 2021']
assert v['all_floor_stock'][0] - v['comparable_stock'][0] == v['upper_floor_cohort'] == 18
display(verdun_inventory)

from render_latex_report_fr import SOURCES
source_rows = [{'Source': author, 'Usage': description.replace('--', '–'), 'URL publique originale': url}
               for author, description, url in SOURCES]
display(pd.DataFrame(source_rows))
for source in study['methodology_sources']['files']:
    display(Markdown(f"**{source['edition']} méthode de l’enquête :** [PDF public original]({source['url']}); "
                     f"[copie vérifiée](../prototype/assets/methodology/{source['path']})."))

from render_latex_report_fr import write_report
report_path = write_report(study, DELIVERABLES / 'montreal_neighbourhood_change_fr.tex')
print(f'Création de {report_path.name} à partir des preuves vérifiées.')
print('Vérifications terminées. Le notebook reprend l’étude achevée.')