Projekt: Funkcyjny pipeline danych Parku Jurajskiego

Nocna zmiana dostaje setki odczytów z sensorów: ruch, napięcie ogrodzeń, temperatura. Część z nich jest uszkodzona, bo brakuje identyfikatora albo wartość jest ujemna, a kierownik chce rano jednego raportu: które strefy są bezpieczne, a dokąd trzeba wysłać ekipę. Jeden wielki skrypt z pętlami i flagami szybko staje się nie do utrzymania.

Czas połączyć wszystkie techniki programowania funkcyjnego w jeden projekt. Zbudujemy funkcyjny system przetwarzania danych dinozaurów - od surowych odczytów sensorów po gotowe raporty bezpieczeństwa. Każdy etap to czysta funkcja, a cały proces to pipeline złożony z pipe i compose.

Architektura projektu

Nasz system składa się z kilku warstw:

1. Parsowanie - wczytanie surowych danych z sensorów
2. Walidacja - sprawdzenie poprawności danych (Maybe/Either)
3. Transformacja - przetworzenie danych (map, filter)
4. Agregacja - podsumowania i statystyki (reduce)
5. Raportowanie - generowanie końcowego raportu

Pracujemy na sześciu odczytach, wśród których celowo ukryliśmy dwie pułapki:

1// Struktura danych wejściowych
2const sensorReadings = [
3  { sensorId: 'S01', zone: 'A', type: 'motion', value: 85, timestamp: 1700000000 },
4  { sensorId: 'S02', zone: 'A', type: 'fence', value: 10000, timestamp: 1700000001 },
5  { sensorId: 'S03', zone: 'B', type: 'motion', value: 12, timestamp: 1700000002 },
6  { sensorId: 'S04', zone: 'B', type: 'temperature', value: 38, timestamp: 1700000003 },
7  { sensorId: null, zone: 'C', type: 'motion', value: -5, timestamp: 1700000004 },
8  { sensorId: 'S06', zone: 'A', type: 'fence', value: 0, timestamp: 1700000005 },
9];

Piąty odczyt nie ma sensorId i ma ujemną wartość - to śmieci, które walidacja musi odrzucić. Szósty to ogrodzenie z napięciem 0: dane poprawne, ale alarmujące. timestamp to czas uniksowy, czyli liczba sekund od 1 stycznia 1970 roku.

Warstwa walidacji z Maybe

Walidator zwraca odczyt albo null, a Maybe z poprzedniej lekcji zamienia null w bezpieczny, pusty pojemnik. Tu wystarczy okrojona wersja z map i getOrElse:

1class Maybe {
2  constructor(value) { this.value = value; }
3  static of(v) { return new Maybe(v); }
4  isNothing() { return this.value === null || this.value === undefined; }
5  map(fn) { return this.isNothing() ? this : Maybe.of(fn(this.value)); }
6  getOrElse(def) { return this.isNothing() ? def : this.value; }
7}
8
9const validateReading = (reading) => {
10  if (!reading.sensorId) return null;
11  if (reading.value < 0) return null;
12  return reading;
13};
14
15const safeValidate = (reading) =>
16  Maybe.of(reading)
17    .map(validateReading)
18    .getOrElse(null);

safeValidate owija odczyt, przepuszcza go przez walidator i od razu odpakowuje: poprawny odczyt wraca bez zmian, a uszkodzony staje się null. Odczyt S06 przechodzi, bo 0 nie jest wartością ujemną. To celowe uproszczenie, bo dalsze etapy pracują na zwykłych tablicach i null odfiltrujemy jednym filter.

Budowanie pipeline

Najpierw trzy małe, czyste funkcje dla pojedynczego odczytu. Każda przyjmuje jeden odczyt i zwraca nowy obiekt, nie ruszając oryginału:

1// Czyste funkcje - każda robi jedną rzecz
2const isValidReading = (r) => r !== null;
3const enrichWithDate = (r) => ({
4  ...r,
5  date: new Date(r.timestamp * 1000).toISOString(),
6});
7const classifyAlert = (r) => ({
8  ...r,
9  alert: r.type === 'fence' && r.value < 5000 ? 'CRITICAL' :
10         r.type === 'motion' && r.value > 80 ? 'WARNING' : 'OK',
11});

enrichWithDate mnoży znacznik przez 1000, bo Date liczy w milisekundach, i zapisuje datę w formacie ISO, np. '2023-11-14T22:13:20.000Z'. classifyAlert używa zagnieżdżonego operatora warunkowego: ogrodzenie poniżej 5000 V to CRITICAL, ruch powyżej 80 to WARNING, a reszta OK.

Dwie kolejne funkcje pracują na całej kolekcji. Grupowanie to znany z lekcji o reduce wzorzec z obliczanym kluczem, a statystyki liczymy dla każdej pary [strefa, odczyty] z Object.entries:

1// Agregacja per strefa
2const groupByZone = (readings) =>
3  readings.reduce((groups, r) => ({
4    ...groups,
5    [r.zone]: [...(groups[r.zone] || []), r],
6  }), {});
7
8// Statystyki per strefa
9const calculateZoneStats = (grouped) =>
10  Object.entries(grouped).map(([zone, readings]) => ({
11    zone,
12    totalReadings: readings.length,
13    alerts: readings.filter((r) => r.alert !== 'OK').length,
14    avgValue: readings.reduce((s, r) => s + r.value, 0) / readings.length,
15    status: readings.some((r) => r.alert === 'CRITICAL') ? 'DANGER' : 'SAFE',
16  }));

some sprawdza, czy choć jeden odczyt w strefie jest krytyczny, i wtedy cała strefa dostaje status DANGER. Uwaga projektowa: avgValue uśrednia razem ruch, wolty i stopnie, co w prawdziwym raporcie nie miałoby sensu. To dobre miejsce na Twoje pierwsze usprawnienie.

Kompletny pipeline z pipe

Teraz składamy wszystko w jedną linię produkcyjną. Kolejność kroków w pipe to dokładnie kolejność ich wykonania:

1function pipe(...fns) {
2  return (value) => fns.reduce((acc, fn) => fn(acc), value);
3}
4
5const generateSecurityReport = pipe(
6  // 1. Walidacja każdego odczytu
7  (readings) => readings.map(safeValidate),
8  // 2. Filtrowanie niepoprawnych
9  (readings) => readings.filter(isValidReading),
10  // 3. Wzbogacanie o datę
11  (readings) => readings.map(enrichWithDate),
12  // 4. Klasyfikacja alertów
13  (readings) => readings.map(classifyAlert),
14  // 5. Grupowanie per strefa
15  groupByZone,
16  // 6. Statystyki
17  calculateZoneStats,
18  // 7. Sortowanie - najniebezpieczniejsze pierwsze
19  (stats) => [...stats].sort((a, b) => b.alerts - a.alerts),
20);
21
22const report = generateSecurityReport(sensorReadings);

Raport ma dwie pozycje. Strefa A: trzy odczyty, dwa alerty (ruch 85 i ogrodzenie 0 V), status DANGER. Strefa B: dwa odczyty, zero alertów, status SAFE. Strefy C nie ma wcale, bo jej jedyny odczyt odpadł na walidacji - w prawdziwym parku milczący sektor to osobny powód do niepokoju. Ostatni krok kopiuje tablicę przed sort, więc nawet sortowanie niczego nie mutuje, a sensorReadings po całym procesie wygląda tak samo jak przed nim.

Rozszerzenie - curried utility functions

Na koniec narzędzia w stylu curried. Konfigurację podajesz najpierw, a tablicę na końcu, więc po skonfigurowaniu każda funkcja pasuje prosto do pipe:

1const filterBy = (key) => (value) => (arr) =>
2  arr.filter((item) => item[key] === value);
3
4const mapField = (key) => (fn) => (arr) =>
5  arr.map((item) => ({ ...item, [key]: fn(item[key]) }));
6
7const filterByZone = filterBy('zone');
8const filterZoneA = filterByZone('A');
9
10const normalizeValues = mapField('value')((v) => Math.round(v * 100) / 100);

filterZoneA to gotowy filtr strefy A, a normalizeValues zaokrągla pole value do dwóch miejsc po przecinku w kopii każdego odczytu. Obie funkcje możesz wstawić jako kolejne etapy generateSecurityReport.

Twoje zadanie dla mentora

W edytorze poniżej znajdziesz kompletny, działający projekt do eksploracji i modyfikacji. Zanim wyślesz link mentorowi, rozbuduj go: zamień Maybe na Either, żeby raport wymieniał odrzucone odczyty z powodem, policz średnie osobno dla każdego typu sensora i oznacz strefy, z których nie dotarł żaden poprawny odczyt. Sprawdź też, czy wejście pozostaje nietknięte: zamroź sensorReadings i każdy odczyt przez Object.freeze, a w trybie ścisłym każda próba mutacji rzuci TypeError. Te same wzorce spotkasz w Świecie 2: reducery w Reduxie to czyste funkcje, a stan w React aktualizuje się niemutowalnymi kopiami.

Pamiętaj: dobry pipeline to linia laboratoryjna z małych, czystych stanowisk - każde robi jedną rzecz, żadne nie rusza oryginalnej próbki, a te same dane zawsze dają ten sam raport.

1// Projekt: Funkcyjny pipeline danych Parku Jurajskiego
2console.log("=== Park Jurajski - Kompletny System Raportowania ===\n");
3
4// --- Maybe monada ---
5class Maybe {
6  constructor(value) { this.value = value; }
7  static of(v) { return new Maybe(v); }
8  isNothing() { return this.value === null || this.value === undefined; }
9  map(fn) { return this.isNothing() ? this : Maybe.of(fn(this.value)); }
10  getOrElse(def) { return this.isNothing() ? def : this.value; }
11}
12
13// --- Utility functions ---
14function pipe(...fns) {
15  return (value) => fns.reduce((acc, fn) => fn(acc), value);
16}
17
18function curry(fn) {
19  return function curried(...args) {
20    if (args.length >= fn.length) return fn(...args);
21    return (...more) => curried(...args, ...more);
22  };
23}
24
25const filterBy = curry((field, value, arr) =>
26  arr.filter((item) => item[field] === value)
27);
28
29// --- Dane wejsciowe (odczyty sensorow) ---
30const sensorReadings = [
31  { sensorId: "S01", zone: "A", type: "motion", value: 85, ts: 1700000000 },
32  { sensorId: "S02", zone: "A", type: "fence", value: 10000, ts: 1700000001 },
33  { sensorId: "S03", zone: "B", type: "motion", value: 12, ts: 1700000002 },
34  { sensorId: "S04", zone: "B", type: "temperature", value: 38, ts: 1700000003 },
35  { sensorId: null, zone: "C", type: "motion", value: -5, ts: 1700000004 },
36  { sensorId: "S06", zone: "A", type: "fence", value: 0, ts: 1700000005 },
37  { sensorId: "S07", zone: "C", type: "motion", value: 92, ts: 1700000006 },
38  { sensorId: "S08", zone: "B", type: "fence", value: 8500, ts: 1700000007 },
39];
40
41// --- Warstwa walidacji ---
42const validateReading = (reading) =>
43  Maybe.of(reading)
44    .map((r) => r.sensorId ? r : null)
45    .map((r) => r && r.value >= 0 ? r : null)
46    .getOrElse(null);
47
48// --- Czyste funkcje transformacji ---
49const isValid = (r) => r !== null;
50
51const enrichWithDate = (r) => ({
52  ...r,
53  date: new Date(r.ts * 1000).toISOString().split("T")[0],
54});
55
56const classifyAlert = (r) => ({
57  ...r,
58  alert: r.type === "fence" && r.value < 5000 ? "CRITICAL"
59       : r.type === "motion" && r.value > 80 ? "WARNING"
60       : "OK",
61});
62
63// --- Agregacja ---
64const groupByZone = (readings) =>
65  readings.reduce((groups, r) => ({
66    ...groups,
67    [r.zone]: [...(groups[r.zone] || []), r],
68  }), {});
69
70const calculateZoneStats = (grouped) =>
71  Object.entries(grouped).map(([zone, readings]) => ({
72    zone,
73    totalReadings: readings.length,
74    criticalAlerts: readings.filter((r) => r.alert === "CRITICAL").length,
75    warnings: readings.filter((r) => r.alert === "WARNING").length,
76    avgValue: Math.round(readings.reduce((s, r) => s + r.value, 0) / readings.length),
77    status: readings.some((r) => r.alert === "CRITICAL") ? "DANGER" : "SAFE",
78  }));
79
80// --- Kompletny pipeline ---
81const generateSecurityReport = pipe(
82  (readings) => readings.map(validateReading),
83  (readings) => readings.filter(isValid),
84  (readings) => readings.map(enrichWithDate),
85  (readings) => readings.map(classifyAlert),
86  groupByZone,
87  calculateZoneStats,
88  (stats) => [...stats].sort((a, b) => b.criticalAlerts - a.criticalAlerts),
89);
90
91// --- Uruchomienie ---
92console.log("Dane wejsciowe:", sensorReadings.length, "odczytow\n");
93
94const report = generateSecurityReport(sensorReadings);
95
96console.log("=== RAPORT BEZPIECZENSTWA ===\n");
97report.forEach((zone) => {
98  const icon = zone.status === "DANGER" ? "!!!" : "OK ";
99  console.log(`[${icon}] Zone ${zone.zone}:`);
100  console.log(`    Odczyty: ${zone.totalReadings}`);
101  console.log(`    Krytyczne: ${zone.criticalAlerts}`);
102  console.log(`    Ostrzezenia: ${zone.warnings}`);
103  console.log(`    Srednia wartosc: ${zone.avgValue}`);
104  console.log(`    Status: ${zone.status}\n`);
105});
106
107// --- Curried filtrowanie raportu ---
108console.log("=== Filtrowane raporty ===\n");
109
110const filterByStatus = filterBy("status");
111const dangerZones = filterByStatus("DANGER")(report);
112console.log("Strefy DANGER:", dangerZones.map((z) => z.zone));
113
114const safeZones = filterByStatus("SAFE")(report);
115console.log("Strefy SAFE:", safeZones.map((z) => z.zone));
116
117// --- Podsumowanie ---
118const summary = report.reduce((acc, z) => ({
119  totalReadings: acc.totalReadings + z.totalReadings,
120  totalCritical: acc.totalCritical + z.criticalAlerts,
121  totalWarnings: acc.totalWarnings + z.warnings,
122}), { totalReadings: 0, totalCritical: 0, totalWarnings: 0 });
123
124console.log("\n=== PODSUMOWANIE ===");
125console.log("Laczne odczyty:", summary.totalReadings);
126console.log("Laczne krytyczne:", summary.totalCritical);
127console.log("Laczne ostrzezenia:", summary.totalWarnings);