"""LearnAI P02 data-loading studio — Python 3, standard library only. Run: python3 data-loader.py --demo python3 data-loader.py --csv measurements.csv --labels targets.csv python3 data-loader.py --check Implement join_targets below, then run --check and add a different-file-shape case of your own. --demo uses an explicitly labelled supplied join, not your implementation. All checks are public practice, not independent certification. No network, installation, personal data or external account is required. Files must use the teaching schema documented on /explore/data-studio/. """ import argparse import csv import hashlib import io import json import math import unittest from pathlib import Path MAX_BYTES = 100_000 MAX_ROWS = 1_000 PARTITIONS = {'training', 'development', 'evaluation'} MEASUREMENTS = 'id,group,split,length,unit\na,A,training,10,mm\nb,B,training,3,cm\nc,C,development,,cm\nd,D,evaluation,80,mm\n' TARGETS = 'target,id\n8,d\n2,c\n1,a\n3,b\n' def finite_number(text): value = float(text) if not math.isfinite(value) or abs(value) > 1_000_000: raise ValueError('Expected a finite number with magnitude at most 1,000,000.') return value def parse_csv(raw, required): """Named columns; arbitrary order; extra columns retained; unique headers. Blank lines are ignored by csv.DictReader. Short/long records are rejected. """ if len(raw) > MAX_BYTES: raise ValueError('File exceeds 100,000 bytes.') reader = csv.DictReader(io.StringIO(raw.decode('utf-8-sig'))) names = reader.fieldnames if not names or len(set(names)) != len(names) or not set(required) <= set(names): raise ValueError('Missing or duplicate column names.') rows = [] for row in reader: if len(rows) >= MAX_ROWS or None in row or any(v is None for v in row.values()): raise ValueError('Too many rows or wrong record width.') rows.append(row) if not rows: raise ValueError('File has no records.') return rows def read_bounded(path): with Path(path).open('rb') as stream: raw = stream.read(MAX_BYTES + 1) if len(raw) > MAX_BYTES: raise ValueError('File exceeds 100,000 bytes.') return raw def measurements(raw): rows = parse_csv(raw, ['id', 'group', 'split', 'length', 'unit']) ids, group_parts, out = set(), {}, [] for row in rows: identity, group, split = row['id'].strip(), row['group'].strip(), row['split'].strip() unit = row['unit'].strip() if not identity or identity in ids or not group or split not in PARTITIONS: raise ValueError('IDs must be unique; groups and declared splits are required.') if unit not in {'cm', 'mm'}: raise ValueError('Unit must be cm or mm, even for missing measurements.') if group in group_parts and group_parts[group] != split: raise ValueError('A measurement group crosses partitions.') ids.add(identity) group_parts[group] = split value = None if not row['length'].strip() else finite_number(row['length']) / (10 if unit == 'mm' else 1) out.append(dict(id=identity, group=group, split=split, length_cm=value)) return out def join_targets(rows, labels): """Learner implementation. Return NEW records in measurement-row order. labels has 'id' and 'target' text columns, possibly shuffled, with extras. Strip IDs. Require nonempty unique IDs and exact ID-set equality: reject missing AND extra label IDs. Parse targets with finite_number. Do not edit rows or labels. Add numeric 'target' to each returned measurement record. Raise ValueError for contract violations. Never pair by row position. """ raise NotImplementedError('Implement the ID join, then run --check.') def supplied_demo_join(rows, labels): """Supplied demonstration only. Exercise implementation remains separate.""" lookup = {} for label in labels: identity = label['id'].strip() if not identity or identity in lookup: raise ValueError('Missing or duplicate target ID.') lookup[identity] = finite_number(label['target']) if set(lookup) != {row['id'] for row in rows}: raise ValueError('Measurement and target IDs differ.') return [dict(row, target=lookup[row['id']]) for row in rows] def summarize(rows, source): """Fit ONLY observed training measurements; apply the same fill elsewhere. Population SD is sqrt(mean squared distance from training mean). Constant training input uses divisor 1, with the constant flag reported explicitly. Labels are deliberately absent from fitting the input transformation. """ observed = [r['length_cm'] for r in rows if r['split'] == 'training' and r['length_cm'] is not None] if not observed: raise ValueError('No observed training measurements; cannot fit preprocessing.') mean = sum(observed) / len(observed) training = [r['length_cm'] if r['length_cm'] is not None else mean for r in rows if r['split'] == 'training'] sd = math.sqrt(sum((x - mean)**2 for x in training) / len(training)) divisor = sd if sd else 1 transformed = [dict(r, filled_cm=mean if r['length_cm'] is None else r['length_cm'], standardized=((mean if r['length_cm'] is None else r['length_cm']) - mean) / divisor) for r in rows] return dict(protocol='named-columns-train-only-v1', source=source, preprocessing=dict(mean_cm=mean, population_sd_cm=sd, divisor_cm=divisor, constant_training_input=sd == 0), partitions={part: sum(r['split'] == part for r in rows) for part in sorted(PARTITIONS)}, rows=transformed, limits='Supplied split membership and group labels are assumptions. No model was fitted or evaluated. Public practice is not independent review.') class JoinPractice(unittest.TestCase): def setUp(self): self.rows = measurements(MEASUREMENTS.encode()) self.labels = parse_csv(TARGETS.encode(), ['id', 'target']) def test_shuffled_join_and_no_mutation(self): before = json.dumps([self.rows, self.labels], sort_keys=True) result = join_targets(self.rows, self.labels) self.assertEqual([r['target'] for r in result], [1, 3, 2, 8]) self.assertEqual([r['id'] for r in result], ['a', 'b', 'c', 'd']) self.assertTrue(all(a is not b for a, b in zip(result, self.rows))) self.assertEqual(json.dumps([self.rows, self.labels], sort_keys=True), before) def test_missing_extra_duplicate_and_nonfinite(self): variants = [self.labels[:-1], self.labels + [dict(id='extra', target='1')], self.labels + [self.labels[0]], [dict(r, target='nan') for r in self.labels]] for bad in variants: with self.subTest(labels=bad), self.assertRaises(ValueError): join_targets(self.rows, bad) def test_different_file_shape(self): labels = parse_csv(b'note,id,target\n"contains, comma",b,30\nnew,a,10\n', ['id', 'target']) result = join_targets(self.rows[:2], labels) self.assertEqual([r['target'] for r in result], [10, 30]) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--demo', action='store_true') parser.add_argument('--check', action='store_true') parser.add_argument('--csv') parser.add_argument('--labels') args = parser.parse_args() if args.check: result = unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(JoinPractice)) raise SystemExit(0 if result.wasSuccessful() else 1) if args.demo: raw, labels_raw = MEASUREMENTS.encode(), TARGETS.encode() join = supplied_demo_join origin = 'Supplied synthetic demonstration; learner join NOT executed.' else: if not args.csv or not args.labels: parser.error('Choose --demo, --check, or both --csv and --labels.') raw, labels_raw = read_bounded(args.csv), read_bounded(args.labels) join = join_targets origin = 'Learner-selected local files. Record their source, license and permitted use separately.' rows = join(measurements(raw), parse_csv(labels_raw, ['id', 'target'])) source = dict(description=origin, measurements_sha256=hashlib.sha256(raw).hexdigest(), targets_sha256=hashlib.sha256(labels_raw).hexdigest(), measurements_bytes=len(raw), targets_bytes=len(labels_raw)) print(json.dumps(summarize(rows, source), indent=2, allow_nan=False)) if __name__ == '__main__': main()