"""LearnAI real-data comparison — Python 3, standard library only. Download this file, iris.csv and the dataset manifest from the project page. The CSV is CC BY 4.0: Fisher, R. (1936). Iris [Dataset]. UCI Machine Learning Repository, https://doi.org/10.24432/C56C76 . Generated row IDs are not specimen IDs. No network calls, package installation or account are required. 1. python3 tabular-project.py --csv iris.csv --baseline 2. Implement fit_line below; add your own tests and retain their output. 3. python3 tabular-project.py --check 4. python3 tabular-project.py --csv iris.csv --develop 5. python3 tabular-project.py --csv iris.csv --freeze choice.json --reason "..." 6. python3 tabular-project.py --csv iris.csv --evaluate choice.json The frozen file is created exclusively: choose a new filename instead of silently overwriting it. Evaluation does not refit on development rows. This is a transparent, editable educational workflow, not secure supervision. Iris appeared in earlier lessons; reserved here does NOT mean previously unseen. Visible checks are development practice, not independent artifact certification. """ import argparse import csv import hashlib import io import json import math from pathlib import Path PROTOCOL = 'iris-petal-from-sepal-grouped-v1' CSV_HASH = 'b34e5958ea7f7625f6a43c060b28d9848e6b810c5d09652b2335ef4619db9df8' EXPOSURE = ('Public Iris measurements were used in preceding lessons. Reserved ' 'within this workflow does not mean previously unseen. Reusing ' 'evaluation outcomes to choose a model makes them development evidence.') SPECIES = ['Iris-setosa', 'Iris-versicolor', 'Iris-virginica'] def fit_line(training): """Learner task: return {'intercept': b, 'slope': m} for y = b + m*x. training is a nonempty list of dictionaries with finite x/y values. Compute mean_x and mean_y using ONLY training rows. Slope is the sum of (x-mean_x)*(y-mean_y), divided by the sum of (x-mean_x)**2. Intercept is mean_y - slope*mean_x. If all x values are equal, use slope 0 and mean_y as intercept. Do not change the input rows. Reject empty, nonfinite or out-of-bound values (absolute value above 100) with ValueError. This formula minimizes training squared error, not MAE. """ raise NotImplementedError('Implement fit_line, then run --check.') def load_split(path): if Path(path).stat().st_size != 5977: raise ValueError('Expected the 5,977-byte source CSV.') raw = Path(path).read_bytes() if len(raw) != 5977 or hashlib.sha256(raw).hexdigest() != CSV_HASH: raise ValueError('Use the unchanged 5,977-byte local Iris CSV linked by LearnAI.') reader = csv.DictReader(io.StringIO(raw.decode('utf-8'))) expected = ['row_id', 'sepal_length_cm', 'sepal_width_cm', 'petal_length_cm', 'petal_width_cm', 'species'] if reader.fieldnames != expected: raise ValueError('Unexpected CSV schema.') records = list(reader) if len(records) != 150: raise ValueError('Expected 150 source rows.') parts = {'training': [], 'development': [], 'evaluation': []} seen, counts = set(), [0, 0, 0] for record in records: vector = tuple(float(record[key]) for key in expected[1:5]) if vector in seen: continue seen.add(vector) species = SPECIES.index(record['species']) bucket = counts[species] % 5 counts[species] += 1 part = 'evaluation' if bucket == 0 else 'development' if bucket == 1 else 'training' parts[part].append(dict(id=record['row_id'], group=vector, species=species, x=vector[0], y=vector[2])) assert [len(parts[k]) for k in parts] == [89, 30, 30] assert len({r['group'] for rows in parts.values() for r in rows}) == 149 return parts def checked_line(training): model = fit_line([dict(row) for row in training]) if (not isinstance(model, dict) or set(model) != {'intercept', 'slope'} or any(isinstance(v, bool) or not isinstance(v, (int, float)) or not math.isfinite(v) or abs(v) > 100 for v in model.values())): raise ValueError('Return two finite bounded line coefficients.') return model def score(rows, model): if not rows: raise ValueError('Cannot score an empty set.') trace = [dict(id=r['id'], x=r['x'], actual=r['y'], predicted=model['intercept'] + model['slope'] * r['x']) for r in rows] errors = [r['actual'] - r['predicted'] for r in trace] return dict(count=len(rows), mae=sum(abs(e) for e in errors)/len(rows), mse=sum(e*e for e in errors)/len(rows), trace=trace) def development(parts, include_line=True): train = parts['training'] models = {'mean': dict(intercept=sum(r['y'] for r in train)/len(train), slope=0)} if include_line: models['line'] = checked_line(train) # Evaluation rows are deliberately absent from comparison output. return {name: dict(model=model, training=score(train, model), development=score(parts['development'], model)) for name, model in models.items()} def frozen_choice(parts, reason): if not reason.strip() or len(reason) > 2000: raise ValueError('A selection reason must contain 1–2,000 characters.') return dict(protocol=PROTOCOL, csv_sha256=CSV_HASH, model=checked_line(parts['training']), training_ids=[r['id'] for r in parts['training']], development_ids=[r['id'] for r in parts['development']], reason=reason.strip(), exposure=EXPOSURE) def evaluate(parts, frozen): if (not isinstance(frozen, dict) or frozen.get('protocol') != PROTOCOL or frozen.get('csv_sha256') != CSV_HASH or frozen.get('training_ids') != [r['id'] for r in parts['training']] or frozen.get('development_ids') != [r['id'] for r in parts['development']]): raise ValueError('Frozen source/protocol/partition does not match.') model = checked_line(parts['training']) if frozen.get('model') != model: raise ValueError('Current fitting code differs from the frozen coefficients. Freeze a new choice explicitly.') return dict(protocol=PROTOCOL, exposure=EXPOSURE, model=model, overall=score(parts['evaluation'], model), species=[dict(label=i, metrics=score([r for r in parts['evaluation'] if r['species'] == i], model)) for i in range(3)]) def visible_checks(): rows = [dict(id=str(i), x=x, y=1+2*x) for i, x in enumerate([0, 1, 2])] before = json.dumps(rows, sort_keys=True) assert fit_line(rows) == dict(intercept=1, slope=2) assert json.dumps(rows, sort_keys=True) == before, 'Do not mutate training records.' assert checked_line([dict(x=2, y=3), dict(x=2, y=5)]) == dict(intercept=4, slope=0) result = score([dict(id='new', x=3, y=8)], checked_line(rows)) assert result['mae'] == 1 and result['mse'] == 1 for invalid in [[], [dict(x=float('nan'), y=2)], [dict(x=1, y=float('inf'))], [dict(x=101, y=2)]]: try: fit_line(invalid) except ValueError: pass else: raise AssertionError('Reject invalid training rows with ValueError.') print('Visible checks passed. Add a new arithmetic and split-isolation test; obtain independent review.') def main(): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument('--csv', default='iris.csv') modes = parser.add_mutually_exclusive_group(required=True) modes.add_argument('--check', action='store_true') modes.add_argument('--baseline', action='store_true') modes.add_argument('--develop', action='store_true') modes.add_argument('--freeze', metavar='NEW_JSON_FILE') modes.add_argument('--evaluate', metavar='FROZEN_JSON_FILE') parser.add_argument('--reason', default='') args = parser.parse_args() if args.check: visible_checks() return parts = load_split(args.csv) if args.freeze: frozen = frozen_choice(parts, args.reason) with Path(args.freeze).open('x', encoding='utf-8') as output: json.dump(frozen, output, indent=2, allow_nan=False) print('Choice frozen. Evaluation remains unrevealed. ' + EXPOSURE) elif args.evaluate: path = Path(args.evaluate) if path.stat().st_size > 20000: raise ValueError('Frozen file exceeds 20 KB.') print(json.dumps(evaluate(parts, json.loads(path.read_text())), indent=2, allow_nan=False)) else: print(json.dumps(dict(protocol=PROTOCOL, exposure=EXPOSURE, comparisons=development(parts, not args.baseline)), indent=2, allow_nan=False)) if __name__ == '__main__': try: main() except (ValueError, NotImplementedError, OSError) as error: raise SystemExit(str(error))