Systém sledování potravin: Blockchain, RFID a IoT
V roce 2013 nalezla laboratoř irského úřadu pro bezpečnost potravin stopy koňské DNA ve výrobcích označených jako „100% hovězí“. Skandál s koňmi se šíří po celé Evropě: 28 zemí zahrnuje více než 900 tun masa staženého z trhu, náklady se odhadují mezi 10 a 14 miliardami eur mezi poškozením obrazu, stažením z trhu a soudním řízením. V roce 2008 melamin falšuje mléko v Číně prášek určený pro novorozence: 300 000 nemocných dětí, 6 potvrzených úmrtí, kolektivní trauma, které ničí důvěru v jídlo na deset let.
Nejde o ojedinělé incidenty. Jsou strukturálním příznakem globálního potravinového řetězce která se každým rokem posouvá dál 8 bilionů dolarů zboží přes tisíce zprostředkovatelé, zpracovatelé, logistici a maloobchodníci s neprůhlednými informacemi, které otevírají dveře podvodům, falšování a kontaminace. Průměrné náklady na stažení potravin v USA dnes přesahují 10 milionů dolarů pouze na přímé náklady (inkaso, likvidace, právní postupy), s výjimkou trvalého poškození dobrého jména. V Evropě hospodaří EFSA průměrně 3 700 oznámení RASFF ročně prostřednictvím systému rychlého varování pro potraviny a krmiva.
Technologická odpověď na tuto krizi důvěry má přesné jméno: komplexní digitální sledovatelnost. Systém, který kombinuje RFID, QR kód, IoT senzory a blockchain k vytvoření neměnné účetní knihy, ověřitelné a konzultovatelné o každé události v životě potravinářského výrobku: z oblasti pěstování do nákupního košíku konečného spotřebitele. Globální trh sledovatelnosti potravin platí dnes 23,3 miliardy dolarů v roce 2025 a vzroste na 44,6 miliardy 2034, poháněné stále přísnějšími předpisy a poptávkou spotřebitelů po transparentnosti bez precedenty.
Tento článek je kompletním technickým průvodcem implementací systému sledovatelnosti potraviny průmyslové kvality. Pokryjeme hardwarovou architekturu (RFID/NFC/QR), komponentu IoT pro monitorování studeného řetězce, povolený blockchain s Hyperledger Fabric, např implementace backendu s Pythonem a FastAPI. Zahrneme případovou studii dodavatelského řetězce z reálného života Italské CHOP a regulační důsledky nařízení EU 178/2002, FSMA 204 a strategie Farm to Fork.
Co se dozvíte v tomto článku
- Komplexní architektura systému sledovatelnosti potravin (z farmy na vidličku)
- RFID/NFC: typy tagů, standardy GS1/EPC, náklady, praktická implementace v Pythonu
- QR Code a GS1 Digital Link pro sledovatelnost tváří v tvář spotřebitelům
- Food Blockchain: Hyperledger Fabric vs Ethereum vs Polygon, klady a zápory
- Chytré smlouvy a řetězový kód pro hromadnou registraci a sledování událostí
- Backend Python/FastAPI: dávkové protokolování, události, genealogické dotazy
- IoT pro studený řetězec: majáky BLE, sledování GPS, automatizovaný HACCP
- GS1 EPCIS 2.0 standardy a předpisy: EU Reg. 178/2002, FSMA 204, ISO 22000
- Případová studie: Parmigiano Reggiano DOP a italská sledovatelnost blockchainu
- ROI, náklady na implementaci a srovnávací tabulka technologií
FoodTech Series – všechny články
| # | Položka | Úroveň | Stát |
|---|---|---|---|
| 1 | IoT Pipeline pro přesné zemědělství s Pythonem a MQTT | Moderní | K dispozici |
| 2 | ML Edge for Crop Disease Detection: TensorFlow Lite na Raspberry Pi | Moderní | K dispozici |
| 3 | Satelitní a meteorologická rozhraní API pro AgriTech: Prediktivní data | Moderní | K dispozici |
| 4 | Systém sledování potravin: Blockchain, RFID a IoT (jste zde) | Moderní | Proud |
| 5 | Počítačová vize pro kontrolu kvality potravin s PyTorch YOLO | Moderní | Již brzy |
| 6 | FSMA 204 Automation: Tracking, Alert and Recall přes Python | Moderní | Již brzy |
| 7 | Automatizace vertikálního zemědělství: Robotické ovládání přes API | Moderní | Již brzy |
| 8 | Prognóza poptávky po redukci odpadu: ML Time-Series | Moderní | Již brzy |
| 9 | Dashboard v reálném čase pro Farm IoT s Angular a Grafana | Střední | Již brzy |
| 10 | Potravinový řetězec: ETL vzor od farmy k maloobchodníkovi | Střední | Již brzy |
Regulační kontext: proč sledovatelnost již není volitelná
Sledovatelnost potravin se za méně než dvacet let změnila z dobrovolné správné praxe na povinnou právně závazné ve všech hlavních jurisdikcích po celém světě. e. porozumět regulačnímu rámci nezbytné před návrhem jakéhokoli technického systému: architektury musí být v souladu by-design, ne dodatečně.
Nařízení EU 178/2002: Evropská nadace
Nařízení ES 178/2002 stanoví obecné zásady evropského potravinového práva a vytváří EFSA (Evropský úřad pro bezpečnost potravin). Článek 18 je základem sledovatelnosti Evropská: ukládá a každý operátory dodavatelského řetězce, aby byli schopni identifikovat každá osoba, která jim poskytla jídlo, krmivo, zvíře nebo jakoukoli látku určené k začlenění do potraviny. Principem je tzv "o krok zpět, jeden krok vpřed": každý operátor musí vědět, od koho dostal a komu dal produkt.
Nařízení nepředepisuje konkrétní technologie: ponechává svobodu provádění, ale ukládá účinnost. To znamená, že systém založený na papíru a byrokracii a formálně vyhovující, ale to se stává nemožným, pokud jde o včasnou reakci na nouzovou situaci. Walmart to empiricky prokázal: před blockchainem sledoval dávku manga vyžadováno 6 dní, 18 hodin a 26 minut. Po, 2,2 sekundy. Tento rozdíl není jen účinnost: je to rozdíl mezi cíleným a hromadným stahováním která ničí celé výrobní kategorie.
FSMA 204: Americký režim
V USA zavedl zákon o modernizaci bezpečnosti potravin (FSMA) z roku 2011 pravidlo 204, který stanoví podrobné požadavky na registraci a sledovatelnost pro zahrnuté produkty v seznamu Food Traceability List (FTL): hlávkový salát, špenát, rajčata, vejce, měkké sýry, rybí produkty, čerstvé řezané ovoce a další vysoce rizikové potraviny.
Aktualizace FSMA 204 – březen 2026
Původní termín plnění 20. ledna 2026 byl prodloužen o 30 měsíců na 20. července 2028 podle zákona o pokračujících prostředcích z roku 2026. FDA uvedl, že nadále podporuje včasnou implementaci a že Firmy by měly začít s modernizací projektů již nyní, aby se vyhnuly přeplněnosti žádostí v posledních měsících. Rozšíření nesnižuje technickou složitost požadavků: Kritické události sledování (CTE) a klíčové datové prvky (KDE) musí být sledovány a přenosné elektronicky za méně než 24 hodin na žádost FDA.
Strategie z farmy na vidličku a digitální produktový pas
Strategie European Farm to Fork (F2F), zveřejněná v roce 2020 jako pilíř zelené dohody, zavádí ambici přinést digitální sledovatelnost na evropskou politickou úroveň. Digital Product Passport (DPP), který je v současné době regulován v této oblasti nařízení o ekodesignu pro udržitelné výrobky (ESPR) stanoví, že do roku 2027 mnoho kategorií potravinářských výrobků musí mít digitální pas ověřitelné obsahující informace o původu, ekologické stopě a spotřebitelském řetězci.
Srovnávací regulační rámec: EU vs. USA vs. zbytek světa
| Jurisdikce | Předpisy | Košťata | Vyžadována technologie | Sankce Max |
|---|---|---|---|---|
| Evropská unie | Reg. 178/2002 + nar. 2019/1381 | Všichni provozovatelé potravin a krmiv | Zdarma (účinné) | Uzavření továrny |
| USA | FSMA 204 | Seznam produktů sledovatelnosti potravin | Elektronický registr | Až 10 milionů USD |
| Čína | Zákon o bezpečnosti potravin z roku 2021 | Dovoz + tuzemsko | Národní platforma sledovatelnosti | Odebrání licence |
| Japonsko | Zákon o označování potravin z roku 2020 | Čerstvé národní produkty | Dobrovolník (pobídka) | Mírný |
| Indie | FSSAI Traceability Reg. 2022 | Vývozci potravin | Systém TRACE je povinný | Až 10 INR |
End-to-End architektura sledovatelnosti: Farm to Fork
Moderní systém sledovatelnosti potravin není jedinou technologickou součástí: a vrstvenou architekturu, která integruje identifikační hardware, sítě IoT, middleware agregace, blockchain jako vrstva důvěry a rozhraní API pro spotřebitele. Každý bod catena generuje události, které musí být zachyceny, ověřeny, archivovány a ověřeny.
6-vrstvá architektura: Farm to Fork
| Úroveň | Herec | Technologie | Vygenerovaná data | Norma |
|---|---|---|---|---|
| 1. Farma | Farmář/chovatel | UHF RFID, IoT senzory, GPS | Položka, původ, zemědělské vstupy, certifikace | GS1 GLN, SGTIN |
| 2.Zpracování | Zpracovatelský závod | RFID HF, čárový kód, systém vidění | Vsádková transformace, přísady, parametry procesu, HACCP | GS1 EPCIS 2.0 |
| 3. Balení | Obal | QR kód GS1 Digital Link, NFC | Datum spotřeby, šarže, složení, alergeny | GS1-128, GTIN |
| 4. Logistika | Dopravník/3PL | BLE maják, GPS tracker, záznamník teplot | Teplota, vlhkost, GPS pozice, doby přepravy | SSCC, GS1 EPCIS |
| 5. Maloobchod | Velkoplošná distribuce | RFID UHF police, POS skener | Příjem zboží, prodej, datum spotřeby, odpad | GS1 EDI, EPCIS |
| 6. Spotřebitel | Konečný spotřebitel | NFC/QR skenování smartphonu | Zkontrolujte pravost, historii produktu, certifikace | GS1 Digital Link URI |
Horizontální vrstva, která prochází všemi úrovněmi a povolený blockchain: každá kritická událost (CTE - Critical Tracking Event) je zapsána v řetězci všemi aktéry, vytvoření neměnného záznamu o životnosti produktu. Podrobné údaje (obrázky, dokumenty, kontinuální měření) zůstávají v systémech operátorů mimo řetězec, ale jejich kryptografický hash je ukotven na řetězu, aby byla zajištěna jeho integrita.
RFID a NFC pro sledování potravin: Kompletní technický průvodce
RFID (Radio Frequency Identification) je páteřní technologií fyzické sledovatelnosti v potravinový dodavatelský řetězec. Na rozdíl od tradičního čárového kódu, RFID nevyžaduje přímou viditelnost, umožňuje současné čtení stovek štítků a lze je číst i přes obal (výjimka: kovové materiály a kapaliny s vysokým obsahem vody, které zhoršují signál UHF až 80 %).
Typy RFID štítků pro potravinářský průmysl
Srovnání identifikačních technologií pro sledovatelnost potravin
| Technologie | Frekvence | Rozsah čtení | Cena štítku | Typické použití | Odolnost |
|---|---|---|---|---|---|
| UHF RFID (Gen2) | 860-960 MHz | 1-10 m | 0,05-0,30 EUR | Palety, balíky, kartony | Standardní (bez kovu/vody) |
| HF RFID (ISO 15693) | 13,56 MHz | 10-100 cm | 0,30-1,50 EUR | Jediný produkt, léky, DOP | Dobré (tekutiny tolerovány) |
| NFC (ISO 14443) | 13,56 MHz | 0-10 cm | 0,20-2,00 EUR | DOP zaměřený na spotřebitele proti padělání | Dobrý (kompatibilní smartphone) |
| BLE majáky | 2,4 GHz | 1-50 m | 5-30 EUR | Chladicí řetěz, paleta s teplotou | Vynikající (integrovaná baterie) |
| QR kód GS1 | N/A (optické) | 5 cm - 5 m | 0,001 EUR (tisk) | Spotřebitelské balení, DL URI | Záleží na potisku |
| p-čip (mikrotranspondér) | Specializované UHF | 0-5 cm | 0,10-0,50 EUR | DOP sýry, prémiové produkty | Vynikající (nerez) |
GS1 Standard pro RFID: SGTIN a EPC
Globálním standardem pro kódování RFID identifikátorů v potravinářském sektoru je GS1 EPC (elektronický kód produktu). Umožňuje kód SGTIN-96 (Serialized Global Trade Item Number). globálně jednoznačně identifikovat každou jednotlivou jednotku produktu:
# Struttura SGTIN-96 (96 bit)
# {Header}.{Filter}.{Partition}.{CompanyPrefix}.{ItemReference}.{SerialNumber}
# Esempio SGTIN per Parmigiano Reggiano DOP
# EPC URN format:
urn:epc:id:sgtin:8012345.067890.123456789
# Decodifica:
# 8012345 = GS1 Company Prefix (assegnato a caseificio specifico)
# 067890 = Item Reference (codice prodotto specifico - forma DOP 24 mesi)
# 123456789 = Serial Number (numero forma univoco)
# Conversione a GTIN-14:
# 0 8012345 067890 [check digit]
# GTIN-14: 08012345067890X
# Python: lettura e parsing SGTIN
import re
from dataclasses import dataclass
@dataclass
class SGTIN:
company_prefix: str
item_reference: str
serial_number: str
gtin: str = ""
def to_urn(self) -> str:
return f"urn:epc:id:sgtin:{self.company_prefix}.{self.item_reference}.{self.serial_number}"
def to_gtin14(self) -> str:
"""Calcola GTIN-14 da company prefix + item reference"""
raw = self.company_prefix + self.item_reference
# Calcola check digit GS1 (Luhn-like algorithm)
total = 0
for i, digit in enumerate(reversed(raw)):
n = int(digit)
if i % 2 == 0:
n *= 3
total += n
check = (10 - (total % 10)) % 10
return f"0{raw}{check}"
def parse_epc_urn(urn: str) -> SGTIN:
"""Parsa un EPC URN SGTIN e restituisce oggetto strutturato"""
pattern = r"urn:epc:id:sgtin:(\d+)\.(\d+)\.(\d+)"
match = re.match(pattern, urn)
if not match:
raise ValueError(f"URN non valido: {urn}")
sgtin = SGTIN(
company_prefix=match.group(1),
item_reference=match.group(2),
serial_number=match.group(3)
)
sgtin.gtin = sgtin.to_gtin14()
return sgtin
# Utilizzo
epc = parse_epc_urn("urn:epc:id:sgtin:8012345.067890.123456789")
print(f"Company: {epc.company_prefix}") # 8012345
print(f"GTIN-14: {epc.gtin}") # 080123450678906
print(f"URN: {epc.to_urn()}")Implementace RFID čtečky s Pythonem
Následující příklad ukazuje integraci s průmyslovou UHF RFID čtečkou (kompatibilní se Zebra FX9600 nebo Impinj Speedway) přes LLRP (Low Level Reader Protocol) nebo rozhraní Proprietární REST:
import asyncio
import json
from datetime import datetime, timezone
from dataclasses import dataclass, field, asdict
from typing import Optional
import httpx
@dataclass
class RFIDEvent:
"""Evento RFID catturato da lettore industriale"""
epc: str # Electronic Product Code
antenna_id: int # ID antenna che ha letto il tag
rssi: float # Signal strength in dBm
timestamp: str # ISO 8601 UTC
reader_id: str # ID lettore (es. "DOCK-01")
location: str # Posizione fisica (es. "INGRESSO-MAGAZZINO")
direction: Optional[str] = None # "IN" | "OUT" | None
@classmethod
def from_reader_data(cls, raw: dict, reader_id: str, location: str) -> "RFIDEvent":
return cls(
epc=raw["epc"],
antenna_id=raw.get("antenna", 1),
rssi=raw.get("rssi", -70.0),
timestamp=datetime.now(timezone.utc).isoformat(),
reader_id=reader_id,
location=location
)
class RFIDEventProcessor:
"""
Processore eventi RFID con deduplicazione e buffer
Gestisce reader industriali via REST API (compatibile Zebra, Impinj)
"""
def __init__(self, reader_url: str, reader_id: str, location: str):
self.reader_url = reader_url
self.reader_id = reader_id
self.location = location
self._seen_epcs: dict[str, float] = {} # EPC -> timestamp last seen
self.dedup_window_seconds = 2.0 # Anti-bounce window
def _is_duplicate(self, epc: str, now: float) -> bool:
"""Filtra letture duplicate nel finestra temporale"""
last_seen = self._seen_epcs.get(epc)
if last_seen and (now - last_seen) < self.dedup_window_seconds:
return True
self._seen_epcs[epc] = now
return False
async def poll_reader(self) -> list[RFIDEvent]:
"""Interroga il lettore RFID e restituisce eventi unici"""
async with httpx.AsyncClient(timeout=5.0) as client:
resp = await client.get(f"{self.reader_url}/api/v1/tags")
resp.raise_for_status()
tags_raw = resp.json().get("tags", [])
now = datetime.now(timezone.utc).timestamp()
events = []
for raw in tags_raw:
epc = raw.get("epc", "")
if not epc or self._is_duplicate(epc, now):
continue
event = RFIDEvent.from_reader_data(raw, self.reader_id, self.location)
events.append(event)
return events
async def stream_events(self, callback, poll_interval: float = 0.5):
"""Stream continuo di eventi RFID con callback"""
print(f"Avvio polling RFID reader {self.reader_id} @ {self.reader_url}")
while True:
try:
events = await self.poll_reader()
for event in events:
await callback(event)
except Exception as e:
print(f"Errore reader {self.reader_id}: {e}")
await asyncio.sleep(poll_interval)
# Utilizzo
async def handle_rfid_event(event: RFIDEvent):
print(f"Lotto {event.epc} rilevato in {event.location} @ {event.timestamp}")
# Invia a pipeline tracciabilita
await send_to_traceability_pipeline(event)
async def send_to_traceability_pipeline(event: RFIDEvent):
async with httpx.AsyncClient() as client:
await client.post(
"http://traceability-api:8000/api/v1/events",
json=asdict(event)
)QR Code a GS1 Digital Link: Sledovatelnost tváří v tvář spotřebitelům
Zatímco RFID je pro koncového spotřebitele neviditelný, QR kód představuje smysl přímý kontakt mezi systémem sledovatelnosti a těmi, kteří si produkt zakoupí. The jednoduché čtení pomocí smartphonu Apple nebo Android se stane oknem do historie kompletní s produktem.
GS1 vyvinula standard GS1 Digital Link URI kódovat v QR kódech nejen GTIN produktu, ale také šarži, datum spotřeby a odkaz direct to online tracking information. The GS1 Digital Link URL has this standardní struktura:
# GS1 Digital Link URI format
# https://{domain}/{primary-key-qualifier}/{value}/{data-qualifier}/{value}?{params}
# Esempio per Parmigiano Reggiano DOP:
# https://trace.parmigianoreggiano.it/01/08012345678905/10/LOT2025001A/17/261231
# Decodifica:
# /01/ = GTIN (Application Identifier AI 01)
# 08012345678905 = GTIN-14 della forma di Parmigiano
# /10/ = Batch/Lot Number (AI 10)
# LOT2025001A = numero lotto specifico
# /17/ = Expiration Date (AI 17)
# 261231 = 31 dicembre 2026
# Python: generazione GS1 Digital Link QR Code
import qrcode
from urllib.parse import urlencode
def generate_gs1_digital_link(
gtin: str, # 14 digits
lot: str, # batch number
expiry: str, # YYMMDD format
domain: str,
serial: str = None
) -> str:
"""
Genera URI GS1 Digital Link compliant (GS1 General Specifications 24.0)
"""
# Validazione GTIN-14
if len(gtin) != 14 or not gtin.isdigit():
raise ValueError(f"GTIN deve essere 14 cifre, ricevuto: {gtin}")
# Build URI path
uri = f"https://{domain}/01/{gtin}/10/{lot}/17/{expiry}"
if serial:
uri += f"/21/{serial}"
return uri
def generate_gs1_qr_code(digital_link_uri: str, output_path: str) -> None:
"""
Genera QR Code GS1 Digital Link con specifiche standard
Versione QR: automatica, ECC Level M (minimo GS1)
"""
qr = qrcode.QRCode(
version=None, # auto-size
error_correction=qrcode.constants.ERROR_CORRECT_M,
box_size=10,
border=4, # quiet zone: min 4 moduli standard GS1
)
qr.add_data(digital_link_uri)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
img.save(output_path)
print(f"QR Code salvato in: {output_path}")
# Esempio utilizzo
gtin_parmigiano = "08012345678905"
lot_number = "PR2025A001"
expiry_date = "261231" # 31 dicembre 2026
uri = generate_gs1_digital_link(
gtin=gtin_parmigiano,
lot=lot_number,
expiry=expiry_date,
domain="trace.parmigianoreggiano.it",
serial="FRM042025001" # numero forma specifico
)
print(f"Digital Link URI: {uri}")
# Output: https://trace.parmigianoreggiano.it/01/08012345678905/10/PR2025A001/17/261231/21/FRM042025001
generate_gs1_qr_code(uri, "/output/parmigiano_qr.png")Blockchain pro sledovatelnost potravin: Architektura a implementace
Blockchain je komponenta, která přináší sledovatelnost z registračního systému to system důvěra. Bez blockchainu (nebo technologického ekvivalentu hlavní knihy distribuované), data sledovatelnosti jsou uložena v centralizovaných řízených systémech jediným aktérem: výrobcem, prodejcem nebo prodejcem softwaru. To znamená že data lze upravit, smazat nebo je jednoduše nesdílet.
Blockchain přináší dvě základní vlastnosti: neměnnost (data zapsaná na řetězci nelze zpětně změnit, aniž by došlo k jejich neplatnosti celý řetězec) e decentralizace trustu (žádný singl herec kontroluje rejstřík: všichni účastníci mají ověřenou kopii).
Hyperledger Fabric vs Ethereum vs Polygon: Který si vybrat?
Srovnání blockchainu pro podnik se sledováním potravin
| Charakteristický | Tkanina Hyperledger | Ethereum (veřejné) | Polygon PoS | VeChain |
|---|---|---|---|---|
| Typ | Soukromé povolení | Veřejnost bez povolení | Veřejná L2 (EVM) | Enterprise povoleno |
| TPS (propustnost) | 3 000–20 000 TPS | 15-30 TPS | 7 000+ TPS | 10 000 TPS |
| Transakční náklady | Infrastruktura (bez plynu) | $1-50+ (volatilní) | 0,001–0,01 USD | 0,0001 $ (VTHO) |
| Ochrana osobních údajů | Vysoká (soukromé kanály) | Žádné (veřejné) | Omezený | Vysoká (soukromé kanály) |
| Vládnutí | konsorcium (MSP) | Decentralizované | Decentralizované | Nadace VeChain |
| Identity účastníků | Certifikáty X.509 (MSP) | Pseudonymní adresy | Pseudonymní adresy | Totožnost ověřena |
| Případová studie potravin | IBM Food Trust, Walmart | Výklenky/startupy | Vznikající | Walmart China, Bright Food |
| Smluvní jazyk | Go, Node.js, Java (řetězový kód) | Pevnost | Pevnost | Pevnost (kompatibilní s EVM) |
| Složitost nastavení | Vysoká (objednatel, MSP, kanály) | Nízký | Nízký | Průměrný |
| Doporučeno pro | Podnik, CHOP, maloobchodní konsorcia | Token, produkt NFT | Startupy, B2C transparentnost | Dodavatelský řetězec Asie, léčiva |
U většiny projektů sledovatelnosti podnikových potravin je na výběr Tkanina Hyperledger: umožňuje vytvořit konsorcium, ve kterém každý Operátor dodavatelského řetězce (výrobce, zpracovatel, logistik, prodejce) se stává „peer“ s certifikovanou identitou, která přesně kontroluje, jaká data sdílí a s kým, prostřednictvím mechanismu kanály a z soukromých sbírek dat.
Řetězový kód Hyperledger Fabric pro sledovatelnost potravin
Řetězový kód Hyperledger Fabric a „chytrá smlouva“. Následující příklad v Go implementuje základní funkce pro sledovatelnost šarže potravin:
// Chaincode Go per Hyperledger Fabric 2.x
// File: food_trace_chaincode.go
package main
import (
"encoding/json"
"fmt"
"time"
"github.com/hyperledger/fabric-contract-api-go/contractapi"
)
// FoodLot rappresenta un lotto alimentare on-chain
type FoodLot struct{
LotID string `json:"lot_id"`
ProductGTIN string `json:"product_gtin"`
ProducerGLN string `json:"producer_gln"`
ProductionDate string `json:"production_date"`
ExpiryDate string `json:"expiry_date"`
CertificateIDs []string `json:"certificate_ids"`
LotStatus string `json:"lot_status"` // ACTIVE | RECALLED | EXPIRED
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
}
// TraceEvent rappresenta un evento di tracciamento sulla supply chain
type TraceEvent struct{
EventID string `json:"event_id"`
LotID string `json:"lot_id"`
EventType string `json:"event_type"` // CREATED | SHIPPED | RECEIVED | TRANSFORMED | SOLD
ActorGLN string `json:"actor_gln"` // GS1 Global Location Number dell'attore
Location string `json:"location"`
Timestamp string `json:"timestamp"`
Temperature *float64 `json:"temperature,omitempty"`
Humidity *float64 `json:"humidity,omitempty"`
Notes string `json:"notes,omitempty"`
DataHash string `json:"data_hash"` // SHA-256 di dati off-chain
}
// FoodTraceContract e il contratto principale
type FoodTraceContract struct{
contractapi.Contract
}
// RegisterLot registra un nuovo lotto alimentare
func (c *FoodTraceContract) RegisterLot(
ctx contractapi.TransactionContextInterface,
lotID, productGTIN, producerGLN, productionDate, expiryDate string,
certificateIDs []string,
) error {
// Verifica che il lotto non esista gia
existing, err := ctx.GetStub().GetState(lotID)
if err != nil {
return fmt.Errorf("errore accesso ledger: %v", err)
}
if existing != nil {
return fmt.Errorf("lotto %s già registrato", lotID)
}
now := time.Now().UTC().Format(time.RFC3339)
lot := FoodLot{
LotID: lotID,
ProductGTIN: productGTIN,
ProducerGLN: producerGLN,
ProductionDate: productionDate,
ExpiryDate: expiryDate,
CertificateIDs: certificateIDs,
LotStatus: "ACTIVE",
CreatedAt: now,
UpdatedAt: now,
}
lotJSON, err := json.Marshal(lot)
if err != nil {
return err
}
return ctx.GetStub().PutState(lotID, lotJSON)
}
// AddTraceEvent aggiunge un evento di tracciamento al lotto
func (c *FoodTraceContract) AddTraceEvent(
ctx contractapi.TransactionContextInterface,
eventJSON string,
) error {
var event TraceEvent
if err := json.Unmarshal([]byte(eventJSON), &event); err != nil {
return fmt.Errorf("JSON evento non valido: %v", err)
}
// Verifica che il lotto esista e sia attivo
lotBytes, err := ctx.GetStub().GetState(event.LotID)
if err != nil || lotBytes == nil {
return fmt.Errorf("lotto %s non trovato", event.LotID)
}
var lot FoodLot
json.Unmarshal(lotBytes, &lot)
if lot.LotStatus != "ACTIVE" {
return fmt.Errorf("lotto non attivo (status: %s)", lot.LotStatus)
}
event.Timestamp = time.Now().UTC().Format(time.RFC3339)
// Chiave composita per eventi: "EVENT" + lotID + eventID
eventKey, err := ctx.GetStub().CreateCompositeKey("EVENT", []string{event.LotID, event.EventID})
if err != nil {
return err
}
eventJSON, err := json.Marshal(event)
if err != nil {
return err
}
return ctx.GetStub().PutState(eventKey, eventJSON)
}
// GetLotHistory restituisce la storia completa di un lotto
func (c *FoodTraceContract) GetLotHistory(
ctx contractapi.TransactionContextInterface,
lotID string,
) ([]TraceEvent, error) {
iterator, err := ctx.GetStub().GetStateByPartialCompositeKey("EVENT", []string{lotID})
if err != nil {
return nil, err
}
defer iterator.Close()
var events []TraceEvent
for iterator.HasNext() {
result, err := iterator.Next()
if err != nil {
continue
}
var event TraceEvent
if err := json.Unmarshal(result.Value, &event); err == nil {
events = append(events, event)
}
}
return events, nil
}
// RecallLot avvia un recall su un lotto
func (c *FoodTraceContract) RecallLot(
ctx contractapi.TransactionContextInterface,
lotID, reason string,
) error {
lotBytes, err := ctx.GetStub().GetState(lotID)
if err != nil || lotBytes == nil {
return fmt.Errorf("lotto %s non trovato", lotID)
}
var lot FoodLot
json.Unmarshal(lotBytes, &lot)
lot.LotStatus = "RECALLED"
lot.UpdatedAt = time.Now().UTC().Format(time.RFC3339)
lotJSON, _ := json.Marshal(lot)
return ctx.GetStub().PutState(lotID, lotJSON)
}Backend Python/FastAPI: Traceability API
Aplikační backend funguje jako middleware mezi terénními systémy (čtečky RFID, IoT) a blockchain. Odhaluje standardizovaná rozhraní REST API pro registraci dávek, sčítání události a dotazy týkající se genealogie produktů. Detailní data jsou archivována v PostgreSQL (s pgvectorem pro budoucí sémantické vyhledávání), zatímco jejich hashe jsou napsané on-chain prostřednictvím Hyperledger Fabric SDK.
# food_traceability_api.py
# FastAPI backend per sistema di tracciabilita alimentare
# Integra PostgreSQL (dati) + Hyperledger Fabric (trust layer)
from fastapi import FastAPI, HTTPException, Depends, status
from pydantic import BaseModel, Field, field_validator
from datetime import datetime, timezone, date
from typing import Optional, List
import hashlib
import json
import uuid
import asyncpg
app = FastAPI(
title="Food Traceability API",
description="Sistema di tracciabilita alimentare end-to-end",
version="2.0.0"
)
# ---- Modelli Pydantic ----
class LotRegistration(BaseModel):
"""Schema per registrazione nuovo lotto alimentare"""
lot_id: str = Field(..., min_length=3, max_length=50, pattern=r"^[A-Z0-9\-]+$")
product_gtin: str = Field(..., min_length=14, max_length=14)
producer_gln: str = Field(..., description="GS1 Global Location Number del produttore")
product_name: str = Field(..., max_length=200)
production_date: date
expiry_date: date
quantity_kg: float = Field(..., gt=0)
origin_country: str = Field(..., max_length=2) # ISO 3166-1 alpha-2
certifications: List[str] = Field(default_factory=list)
raw_materials: List[dict] = Field(default_factory=list)
@field_validator("product_gtin")
@classmethod
def validate_gtin14(cls, v: str) -> str:
"""Valida GTIN-14 con GS1 check digit algorithm"""
if not v.isdigit():
raise ValueError("GTIN deve contenere solo cifre")
digits = [int(d) for d in v]
total = sum(
d * (3 if (len(digits) - 1 - i) % 2 == 0 else 1)
for i, d in enumerate(digits[:-1])
)
expected_check = (10 - (total % 10)) % 10
if digits[-1] != expected_check:
raise ValueError(f"Check digit GTIN non valido. Atteso: {expected_check}")
return v
@field_validator("expiry_date")
@classmethod
def expiry_after_production(cls, v, info):
if "production_date" in info.data and v <= info.data["production_date"]:
raise ValueError("Data scadenza deve essere successiva alla data produzione")
return v
class TraceEventRequest(BaseModel):
"""Schema per aggiunta evento di tracciamento"""
lot_id: str
event_type: str = Field(..., pattern=r"^(CREATED|SHIPPED|RECEIVED|TRANSFORMED|STORED|SOLD|RECALLED)$")
actor_gln: str
actor_name: str
location_name: str
location_gln: Optional[str] = None
latitude: Optional[float] = None
longitude: Optional[float] = None
temperature_celsius: Optional[float] = None
humidity_percent: Optional[float] = None
quantity_kg: Optional[float] = None
notes: Optional[str] = None
additional_data: dict = Field(default_factory=dict)
class LotGenealogy(BaseModel):
"""Genealogia completa di un lotto"""
lot_id: str
product_name: str
product_gtin: str
producer_name: str
production_date: str
expiry_date: str
status: str
certifications: List[str]
events: List[dict]
blockchain_tx_hash: Optional[str] = None
raw_materials: List[dict]
# ---- Endpoints ----
@app.post("/api/v1/lots", status_code=status.HTTP_201_CREATED)
async def register_lot(lot: LotRegistration, db=Depends(get_db)):
"""
Registra un nuovo lotto alimentare.
Scrive il record in PostgreSQL e l'hash on-chain su Hyperledger Fabric.
"""
# Verifica unicita lotto
existing = await db.fetchrow(
"SELECT lot_id FROM food_lots WHERE lot_id = $1", lot.lot_id
)
if existing:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"Lotto {lot.lot_id} già registrato"
)
# Calcola hash dei dati per ancoraggio blockchain
lot_data = lot.model_dump(mode="json")
lot_json_str = json.dumps(lot_data, sort_keys=True, default=str)
data_hash = hashlib.sha256(lot_json_str.encode()).hexdigest()
# Persisti su PostgreSQL
await db.execute("""
INSERT INTO food_lots (
lot_id, product_gtin, producer_gln, product_name,
production_date, expiry_date, quantity_kg, origin_country,
certifications, raw_materials, lot_status, data_hash, created_at
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10::jsonb, 'ACTIVE', $11, NOW())
""",
lot.lot_id, lot.product_gtin, lot.producer_gln, lot.product_name,
lot.production_date, lot.expiry_date, lot.quantity_kg, lot.origin_country,
json.dumps(lot.certifications), json.dumps(lot.raw_materials), data_hash
)
# Scrivi hash su blockchain (asincrono)
tx_hash = await write_to_blockchain(lot.lot_id, data_hash, "REGISTER_LOT")
return {
"lot_id": lot.lot_id,
"status": "registered",
"data_hash": data_hash,
"blockchain_tx": tx_hash,
"message": f"Lotto {lot.lot_id} registrato con successo"
}
@app.post("/api/v1/lots/{lot_id}/events", status_code=status.HTTP_201_CREATED)
async def add_trace_event(lot_id: str, event: TraceEventRequest, db=Depends(get_db)):
"""Aggiunge un evento di tracciamento a un lotto esistente"""
# Verifica lotto esiste ed e attivo
lot = await db.fetchrow(
"SELECT lot_id, lot_status FROM food_lots WHERE lot_id = $1", lot_id
)
if not lot:
raise HTTPException(status_code=404, detail=f"Lotto {lot_id} non trovato")
if lot["lot_status"] not in ("ACTIVE", "STORED"):
raise HTTPException(
status_code=400,
detail=f"Lotto in stato {lot['lot_status']}: non accetta nuovi eventi"
)
event_id = str(uuid.uuid4())
timestamp = datetime.now(timezone.utc).isoformat()
# Hash dati evento per blockchain
event_data = event.model_dump(mode="json")
event_data["event_id"] = event_id
event_data["timestamp"] = timestamp
event_json = json.dumps(event_data, sort_keys=True, default=str)
event_hash = hashlib.sha256(event_json.encode()).hexdigest()
# Persisti evento
await db.execute("""
INSERT INTO trace_events (
event_id, lot_id, event_type, actor_gln, actor_name,
location_name, location_gln, latitude, longitude,
temperature_celsius, humidity_percent, quantity_kg,
notes, additional_data, data_hash, created_at
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14::jsonb,$15,NOW())
""",
event_id, lot_id, event.event_type, event.actor_gln, event.actor_name,
event.location_name, event.location_gln, event.latitude, event.longitude,
event.temperature_celsius, event.humidity_percent, event.quantity_kg,
event.notes, json.dumps(event.additional_data), event_hash
)
# Aggiorna status lotto
status_map = {"SHIPPED": "IN_TRANSIT", "RECEIVED": "STORED", "SOLD": "SOLD"}
if event.event_type in status_map:
await db.execute(
"UPDATE food_lots SET lot_status = $1 WHERE lot_id = $2",
status_map[event.event_type], lot_id
)
tx_hash = await write_to_blockchain(lot_id, event_hash, event.event_type)
return {"event_id": event_id, "blockchain_tx": tx_hash, "status": "recorded"}
@app.get("/api/v1/lots/{lot_id}/genealogy", response_model=LotGenealogy)
async def get_lot_genealogy(lot_id: str, db=Depends(get_db)):
"""Restituisce la genealogia completa di un lotto: dati base + tutti gli eventi"""
lot = await db.fetchrow("""
SELECT l.*, p.name as producer_name
FROM food_lots l
LEFT JOIN producers p ON l.producer_gln = p.gln
WHERE l.lot_id = $1
""", lot_id)
if not lot:
raise HTTPException(status_code=404, detail=f"Lotto {lot_id} non trovato")
events = await db.fetch("""
SELECT * FROM trace_events
WHERE lot_id = $1
ORDER BY created_at ASC
""", lot_id)
blockchain_tx = await get_blockchain_anchor(lot_id)
return LotGenealogy(
lot_id=lot["lot_id"],
product_name=lot["product_name"],
product_gtin=lot["product_gtin"],
producer_name=lot.get("producer_name", lot["producer_gln"]),
production_date=str(lot["production_date"]),
expiry_date=str(lot["expiry_date"]),
status=lot["lot_status"],
certifications=json.loads(lot["certifications"] or "[]"),
events=[dict(e) for e in events],
blockchain_tx_hash=blockchain_tx,
raw_materials=json.loads(lot["raw_materials"] or "[]")
)
async def write_to_blockchain(lot_id: str, data_hash: str, event_type: str) -> str:
"""
Scrive un hash su Hyperledger Fabric via REST API (Fabric Gateway API)
In produzione: usa fabric-sdk-py o Fabric Gateway REST proxy
"""
# Stub: in produzione integra con fabric-gateway
tx_id = hashlib.sha256(f"{lot_id}{data_hash}{event_type}".encode()).hexdigest()
return f"0x{tx_id}"
async def get_db():
"""Dependency: connessione PostgreSQL via asyncpg"""
conn = await asyncpg.connect("postgresql://user:pass@localhost/traceability")
try:
yield conn
finally:
await conn.close()IoT for Cold Chain: Monitorování teploty a automatizovaný HACCP
Chladicí řetězec je kritickým bodem bezpečnosti potravin mléčné výrobky, maso, ryby a čerstvé produkty. 20-30 % přetržení studeného řetězu dojde během přepravy a jakákoli odchylka teploty mimo rozsah může určit exponenciální množení bakterií a zdravotní rizika. Systém IoT integrovaný umožňuje nepřetržité monitorování s automatickými výstrahami a nahráváním v souladu s požadavky HACCP (analýza rizik a kritické kontrolní body).
Cold Chain IoT architektura
IoT Technology Stack pro monitorování studeného řetězce
| Komponent | Technologie | Protokol | Orientační cena |
|---|---|---|---|
| Senzor teploty/vlhkosti | Ruuvi Tag, Minew S1, Onset HOBO | BLE 4.2/5.0 | 15-80 EUR/kus |
| Brána BLE-to-Cloud | Raspberry Pi 4 + BLE dongle | MQTT přes 4G/WiFi | 80-200 EUR |
| GPS sledovač palet | Teltonika FMB920, Queclink GL320MG | LTE-M, MQTT | 60-150 EUR |
| Samostatný záznamník dat | Tec4med D2, Berlinger FRIDGE-tag | USB/NFC stahování | 30-120 EUR |
| Databáze časových řad | InfluxDB 3.x, TimescaleDB | Protokol HTTP/Line | Open source / Cloud |
| Upozornění a oznámení | Grafana Alerting, PagerDuty | Webhooky, e-mail, SMS | Záleží na objemu |
# cold_chain_monitor.py
# Monitor IoT per cold chain con BLE scanning e MQTT publishing
# Compatible con Ruuvi Tag v2 (formato RAWv2)
import asyncio
import json
import struct
from datetime import datetime, timezone
from dataclasses import dataclass, asdict
from typing import Optional
import paho.mqtt.client as mqtt
# Configurazione limiti HACCP per categoria prodotto
HACCP_LIMITS = {
"dairy": {"min": 0.0, "max": 4.0, "alert_threshold": 6.0},
"fresh_meat": {"min": -2.0, "max": 4.0, "alert_threshold": 5.0},
"frozen": {"min": -25.0, "max": -18.0, "alert_threshold": -15.0},
"fish": {"min": -1.0, "max": 2.0, "alert_threshold": 4.0},
"vegetables": {"min": 2.0, "max": 8.0, "alert_threshold": 10.0},
}
@dataclass
class ColdChainReading:
"""Lettura sensore cold chain"""
sensor_mac: str
lot_id: str
timestamp: str
temperature_c: float
humidity_pct: float
battery_pct: int
latitude: Optional[float] = None
longitude: Optional[float] = None
product_category: str = "dairy"
alert: bool = False
alert_reason: str = ""
def check_haccp_compliance(self) -> None:
"""Verifica compliance HACCP e setta flag alert"""
limits = HACCP_LIMITS.get(self.product_category, HACCP_LIMITS["dairy"])
if self.temperature_c > limits["alert_threshold"]:
self.alert = True
self.alert_reason = (
f"TEMPERATURA CRITICA: {self.temperature_c:.1f}C "
f"(limite: {limits['alert_threshold']}C)"
)
elif self.temperature_c < limits["min"]:
self.alert = True
self.alert_reason = (
f"TEMPERATURA SOTTO MINIMO: {self.temperature_c:.1f}C "
f"(minimo: {limits['min']}C)"
)
elif self.temperature_c > limits["max"]:
# Fuori range ma non ancora critico
self.alert_reason = (
f"Temperatura fuori range HACCP: {self.temperature_c:.1f}C "
f"(range: {limits['min']}-{limits['max']}C)"
)
def parse_ruuvi_rawv2(manufacturer_data: bytes) -> dict:
"""
Parsa il payload RAWv2 di Ruuvi Tag (formato 0x05)
Documentazione: https://docs.ruuvi.com/communication/bluetooth-advertisements/data-format-5-rawv2
"""
if len(manufacturer_data) < 24 or manufacturer_data[0] != 0x05:
raise ValueError("Formato Ruuvi RAWv2 non valido")
# Unpack: temperatura (int16, 0.005 C/unit), umidita (uint16, 0.0025 %/unit)
temp_raw = struct.unpack_from(">h", manufacturer_data, 1)[0]
hum_raw = struct.unpack_from(">H", manufacturer_data, 3)[0]
batt_raw = struct.unpack_from(">H", manufacturer_data, 12)[0]
temperature = temp_raw * 0.005
humidity = hum_raw * 0.0025
battery_mv = ((batt_raw >> 5) + 1600) # mV
battery_pct = max(0, min(100, int((battery_mv - 2200) / 12)))
return {
"temperature_c": round(temperature, 2),
"humidity_pct": round(humidity, 1),
"battery_pct": battery_pct,
}
class ColdChainMQTTPublisher:
"""Pubblica letture cold chain su MQTT broker"""
def __init__(self, broker_host: str, broker_port: int = 1883):
self.client = mqtt.Client(client_id="cold-chain-monitor-01")
self.client.connect(broker_host, broker_port, keepalive=60)
self.client.loop_start()
def publish_reading(self, reading: ColdChainReading) -> None:
topic = f"coldchain/{reading.lot_id}/sensors/{reading.sensor_mac}"
payload = json.dumps(asdict(reading), default=str)
self.client.publish(topic, payload, qos=1, retain=False)
if reading.alert:
alert_topic = f"coldchain/alerts/{reading.lot_id}"
self.client.publish(alert_topic, payload, qos=2)
print(f"ALERT {reading.lot_id}: {reading.alert_reason}")
async def cold_chain_pipeline(publisher: ColdChainMQTTPublisher, lot_id: str):
"""
Pipeline simulata - in produzione: integra con BLE scanner (bleak library)
e gateway che forwarda dati da sensori fisici
"""
import random
import time
sensor_mac = "AA:BB:CC:DD:EE:FF"
product_category = "dairy"
while True:
# Simulazione lettura sensore (sostituire con bleak BLE scan)
simulated_temp = round(random.gauss(3.5, 1.2), 2) # Media 3.5C, std 1.2
simulated_hum = round(random.gauss(85.0, 5.0), 1)
reading = ColdChainReading(
sensor_mac=sensor_mac,
lot_id=lot_id,
timestamp=datetime.now(timezone.utc).isoformat(),
temperature_c=simulated_temp,
humidity_pct=simulated_hum,
battery_pct=85,
product_category=product_category
)
reading.check_haccp_compliance()
publisher.publish_reading(reading)
print(f"{reading.timestamp} | Temp: {reading.temperature_c}C | "
f"Hum: {reading.humidity_pct}% | Alert: {reading.alert}")
await asyncio.sleep(30) # Lettura ogni 30 secondi
# Avvio pipeline
# publisher = ColdChainMQTTPublisher("mqtt.traceability.local")
# asyncio.run(cold_chain_pipeline(publisher, "LOT-PR2025-001"))GS1 EPCIS 2.0 Standard: Společný jazyk sledovatelnosti
EPCIS (Electronic Product Code Information Services) a standard GS1, který definuje jak by měly být události sledovatelnosti strukturovány a sdíleny mezi systémy heterogenní. Verze 2.0, publikovaná v roce 2022 a aktualizovaná podle pokynů FSMA 204 v květnu 2025, představuje radikální přepsání, které přináší sledovatelnost dodávek řetězec v moderní webové éře s nativní podporou JSON/JSON-LD a REST API.
5 typů událostí EPCIS 2.0
Typy událostí EPCIS 2.0 pro sledovatelnost potravin
| Událost | Kdy jej použít | Příklad jídla | CTE FSMA 204 |
|---|---|---|---|
| ObjectEvent | Akce na jednom objektu/pozici | Sériová výroba, příjem, expedice | Pěstování, příjem, doprava |
| AgregationEvent | Agregace (paleta, kontejner) | Balení sýrů CHOP do palet, kontejnerizace | Doprava (balení) |
| TransactionEvent | Obchodní transakce | Objednávka, faktura, dodací list | N/A (komerční dodavatelský řetězec) |
| TransformationEvent | Dávková transformace na nové produkty | Čerstvé mléko vyrobené do sýra | Transformace |
| AssociationEvent | Asociace mezi objekty | Přidružení senzoru IoT k dávce | N/A |
# Esempio EPCIS 2.0 JSON-LD - Evento di spedizione lotto DOP
# Formato JSON standard GS1 EPCIS 2.0 (May 2025 update)
epcis_shipping_event = {
"@context": [
"https://ref.gs1.org/standards/epcis/2.0.0/epcis-context.jsonld",
{
"ext": "https://parmigianoreggiano.it/epcis/ext/"
}
],
"type": "EPCISDocument",
"schemaVersion": "2.0",
"creationDate": "2025-10-15T08:30:00.000Z",
"epcisBody": {
"eventList": [
{
"type": "ObjectEvent",
"eventTime": "2025-10-15T08:00:00.000Z",
"eventTimeZoneOffset": "+01:00",
"action": "OBSERVE",
"bizStep": "shipping", # GS1 CBV vocabulary
"disposition": "in_transit",
"epcList": [
"urn:epc:id:sgtin:8012345.067890.001",
"urn:epc:id:sgtin:8012345.067890.002",
"urn:epc:id:sgtin:8012345.067890.003"
],
"readPoint": {
"id": "urn:epc:id:sgln:8012345.00000.DOCK-01" # GLN dock uscita
},
"bizLocation": {
"id": "urn:epc:id:sgln:8012345.00000.WAREHOUSE"
},
"bizTransactionList": [
{
"type": "po", # Purchase Order
"bizTransaction": "urn:epcglobal:cbv:bt:9876543210:PO-2025-1234"
},
{
"type": "desadv", # Dispatch Advice (DDT)
"bizTransaction": "urn:epcglobal:cbv:bt:9876543210:DDT-2025-5678"
}
],
"sourceList": [
{
"type": "owning_party",
"source": "urn:epc:id:pgln:8012345.00000" # Consorzio PR
}
],
"destinationList": [
{
"type": "owning_party",
"destination": "urn:epc:id:pgln:4099999.00000" # Retailer
}
],
"sensorElementList": [
{
"sensorMetadata": {
"time": "2025-10-15T08:00:00.000Z",
"deviceID": "urn:epc:id:giai:8012345.0.SENSOR-TEMP-01",
"deviceMetadata": "https://sensors.parmigianoreggiano.it/models/T1"
},
"sensorReport": [
{
"type": "Temperature",
"value": 3.8,
"uom": "CEL", # Celsius (UN/CEFACT unit code)
"minValue": 3.2,
"maxValue": 4.1
},
{
"type": "AbsoluteHumidity",
"value": 82.5,
"uom": "A93" # Percent relative humidity
}
]
}
],
"ext:lotNumber": "PR2025A001",
"ext:dop_certificate": "DOP-IT-PR-2025-001234"
}
]
}
}
# Invio a EPCIS 2.0 Repository (es. IBM Food Trust, OPTEL, TraceLink)
import httpx
async def publish_epcis_event(event: dict, epcis_endpoint: str, api_key: str):
"""Pubblica evento EPCIS 2.0 su repository conforme"""
async with httpx.AsyncClient() as client:
resp = await client.post(
f"{epcis_endpoint}/events",
json=event,
headers={
"Content-Type": "application/ld+json",
"GS1-EPCIS-Version": "2.0",
"Authorization": f"Bearer {api_key}"
}
)
resp.raise_for_status()
return resp.json()Případová studie: Parmigiano Reggiano DOP a sledovatelnost blockchainu
Konsorcium Parmigiano Reggiano představuje nejpokročilejší případovou studii digitální sledovatelnost v italském dodavatelském řetězci DOP. S více než 3 600 chovy dobytka, 311 produkujících mlékáren, 1 200 koření a hodnota produkce 3 miliardy eur ročně (2024), výzva sledovatelnosti a průmyslovém měřítku s právně závaznými požadavky na pravost.
Systém p-Chip: Fyzický blockchain ve formách
Od roku 2022 zahájilo Konsorcium program s Kaasmerk Matec e p-Chip Corporation integrovat do historické kaseinové desky (identifikační systém používaný od roku 2002) kryptografický mikrotranspondér. P-Chip je menší než zrnko soli a je odolný vůči teplotám od -40C do +300C, na sýrové kyseliny a na dlouhé cykly zrání až 36 měsíců. Každý čip obsahuje jedinečný kryptografický identifikátor, který je fixován do účetní knihy blockchainu, čímž se vytvoří a "digitální dvojče" tvaru.
Systém sledovatelnosti architektury Parmigiano Reggiano DOP
| Fáze | Technologie | Registrovaná data | Herci |
|---|---|---|---|
| Hospodařit | BDN (národní databáze skotu), IoT | Identifikace skotu, výživa, welfare zvířat, obec původu mléka | Konsorcium registrovaní chovatelé |
| Mléko | p-Chip aplikovaný na kaseinovou desku, HF RFID | ID tvaru, datum výroby, mlékárna, sériové číslo MiPAAF, množství mléka, parametry HACCP | 311 DOP mlékárny |
| Zkouška kvality | Fire branding + ověření p-Chipu | Složená odborná zkouška (12 měsíců), stupeň kvality, známka DOP | Expert konsorcia |
| Koření | Senzory teploty/vlhkosti, RFID sledování | Teplota, vlhkost, doba zrání, pohyby | 1200 koření |
| Porcování | QR kód GS1 Digital Link na balení | Šarže, tvar původu, datum balení, sledovatelnost DOP | Autorizovaní porcovatelé |
| Spotřebitel | Naskenujte QR/NFC pomocí smartphonu | Celý příběh: chov, mlékárna, zrání | Konečný spotřebitel |
ROI a ekonomický dopad
Výsledky programu digitální sledovatelnosti Parmigiano Reggiano DOP
- Omezení padělání: "Italian Sounding" (produkty napodobující PR mimo oblast CHOP) stojí konsorcium ~2,2 miliardy EUR/rok ve ztrátě obratu. Digitální sledovatelnost umožňuje ověřit pravost v jakémkoli bodě globálního distribučního řetězce, zejména v USA, Kanadě a Austrálii, kde je Italian Sounding nejrozšířenější.
- Časy připomenutí: V případě zdravotní výstrahy systém zkracuje dobu pro identifikaci rizikových šarží z 3-5 dnů (s papírovými záznamy) na méně než 2 hodiny.
- Cena prémie: Ověřitelná sledovatelnost umožňuje ospravedlnit prémiové ceny ve výši 15–25 % na mezinárodních trzích, kde má spotřebitel nástroje k ověření pravosti.
- Přístup na trh: Digitální soulad usnadňuje přístup k mezinárodnímu maloobchodu ve velkém měřítku (Whole Foods, Waitrose, Monoprix), který stále více vyžaduje digitální dokumentaci dodavatelského řetězce.
- Náklady na implementaci: Odhaduje se 8-15 EUR na formu v prvních letech (capex + opex), s očekávanou návratností během 3-4 let díky výhodám prémiových cen a snížení počtu stažení.
Anti-Patterns and Risks: The Truth about the Food Blockchain
Nadšení pro blockchain ve sledovatelnosti potravin vyvolalo mnoho projektů předimenzované, špatně navržené nebo přímo selhání. IBM Food Trust, produkt reference založená na Hyperledger Fabric, oznámila ukončení služby v prosinci 2022 (poté prodloužena), což odhalilo potíže obchodního modelu v blockchainovém potravinovém ekosystému. Pochopení anti-vzorců je základem navrhování systémů, které skutečně fungují.
7 anti-vzorců blockchainové sledovatelnosti potravin
- Garbage In, Garbage Out (nejnebezpečnější): Blockchain zaručuje neměnnost zadaných údajů, nikoli jejich správnost. Pokud nepoctivý operátor vloží nepravdivé údaje (BIO certifikovaný produkt, který není), blockchain je certifikuje neměnně jako pravdivé. Sledovatelnost bez ověření na místě nebo IoT senzorů nezávislá a iluzorní ochrana.
- Blockchain jako databáze: Použití blockchainu k ukládání dat kompletní s sledovatelností (obrázky, dokumenty, průběžná měření) a neefektivní a drahé. Blockchain musí obsahovat pouze minimální kryptografické hashe a metadata. Skutečná data leží v tradičních databázích.
- Přecenění složitosti konsensu: Mnoho společností má implementoval veřejný blockchain (Ethereum) pro potravinové dodavatelské řetězce, platící nepředvídatelné poplatky za plyn a trpící latencí u každé operace. Pro dodavatelský řetězec jídlo, povolené blockchainy (Fabric, Quorum) jsou téměř vždy správnou volbou.
- Ignorování interoperability: Samotný interní blockchain má malou cenu. Hodnota se objeví, když všichni aktéři v dodavatelském řetězci (výrobce, logistik, maloobchodník) sdílejí stejnou účetní knihu. Bez otevřených standardů (EPCIS 2.0) vznikají ostrovy drahé digitální fotoaparáty.
- Podcenění vstupních nákladů: Skutečná cena není blockchain sama, ale přivede 50–100 dodavatelů EMS (často farmy s omezeným digitalizace) na platformě. Konzervativní rozpočet: 1 500-5 000 EUR na osobu dodavatele v oblasti školení a integrace.
- Nedostatek správy konsorcia: Kdo spravuje uzly? Kdo schvaluje noví účastníci? Kdo platí infrastrukturu? Bez jasného vládnutí blockchain konsorcium, projekty uvíznou na dynamice moci mezi aktéry konkurenty ve stejném dodavatelském řetězci.
- Zapomínání na spotřebitele: Sledovatelnost existuje za účelem vytváření hodnoty. Pokud spotřebitel nemá snadný přístup k informacím (nečitelný QR kód, pomalá webová aplikace, nesrozumitelná technická data), investice se nepromítne do tržní prémie.
Schéma databáze PostgreSQL pro systém sledovatelnosti
-- Schema PostgreSQL per sistema di tracciabilita alimentare
-- Compatibile con EPCIS 2.0 e FSMA 204 CTE/KDE requirements
-- Estensione per UUID e JSONB
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
CREATE EXTENSION IF NOT EXISTS pgcrypto;
-- Tabella produttori (GS1 GLN)
CREATE TABLE producers (
gln VARCHAR(13) PRIMARY KEY, -- GS1 Global Location Number
name VARCHAR(200) NOT NULL,
country CHAR(2) NOT NULL, -- ISO 3166-1 alpha-2
region VARCHAR(100),
certifications JSONB DEFAULT '[]',
contact_email VARCHAR(200),
created_at TIMESTAMPTZ DEFAULT NOW()
);
-- Tabella lotti alimentari (core entity)
CREATE TABLE food_lots (
lot_id VARCHAR(50) PRIMARY KEY,
product_gtin CHAR(14) NOT NULL,
producer_gln VARCHAR(13) REFERENCES producers(gln),
product_name VARCHAR(200) NOT NULL,
production_date DATE NOT NULL,
expiry_date DATE NOT NULL,
quantity_kg DECIMAL(12, 3),
origin_country CHAR(2),
certifications JSONB DEFAULT '[]',
raw_materials JSONB DEFAULT '[]', -- Array di lot_id input + quantità
lot_status VARCHAR(20) DEFAULT 'ACTIVE'
CHECK (lot_status IN ('ACTIVE','IN_TRANSIT','STORED','SOLD','RECALLED','EXPIRED')),
data_hash CHAR(64), -- SHA-256 dei dati per ancoraggio blockchain
blockchain_tx VARCHAR(66), -- Hash TX blockchain (0x + 64 hex)
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
CONSTRAINT valid_dates CHECK (expiry_date > production_date)
);
-- Indici per query frequenti
CREATE INDEX idx_food_lots_gtin ON food_lots(product_gtin);
CREATE INDEX idx_food_lots_producer ON food_lots(producer_gln);
CREATE INDEX idx_food_lots_status ON food_lots(lot_status);
CREATE INDEX idx_food_lots_production_date ON food_lots(production_date);
-- Tabella eventi di tracciamento (Critical Tracking Events - FSMA 204)
CREATE TABLE trace_events (
event_id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
lot_id VARCHAR(50) NOT NULL REFERENCES food_lots(lot_id) ON DELETE RESTRICT,
event_type VARCHAR(30) NOT NULL
CHECK (event_type IN (
'CREATED','SHIPPED','RECEIVED','TRANSFORMED',
'STORED','SOLD','RECALLED','SAMPLED','INSPECTED'
)),
-- Key Data Elements (KDE) FSMA 204
actor_gln VARCHAR(13),
actor_name VARCHAR(200) NOT NULL,
location_name VARCHAR(200) NOT NULL,
location_gln VARCHAR(13),
latitude DECIMAL(9, 6),
longitude DECIMAL(9, 6),
-- Parametri ambientali (cold chain / HACCP)
temperature_celsius DECIMAL(5, 2),
humidity_percent DECIMAL(5, 2),
quantity_kg DECIMAL(12, 3),
-- Documenti associati (DDT, fatture, certificati)
document_refs JSONB DEFAULT '[]',
-- Dati liberi addizionali
additional_data JSONB DEFAULT '{}',
notes TEXT,
data_hash CHAR(64),
blockchain_tx VARCHAR(66),
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX idx_trace_events_lot_id ON trace_events(lot_id);
CREATE INDEX idx_trace_events_type ON trace_events(event_type);
CREATE INDEX idx_trace_events_created ON trace_events(created_at);
CREATE INDEX idx_trace_events_location ON trace_events(location_gln);
-- Tabella alert HACCP automatici
CREATE TABLE haccp_alerts (
alert_id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
lot_id VARCHAR(50) REFERENCES food_lots(lot_id),
sensor_id VARCHAR(100),
alert_type VARCHAR(50) NOT NULL, -- TEMP_HIGH, TEMP_LOW, HUMIDITY, etc.
measured_value DECIMAL(8, 3),
threshold_value DECIMAL(8, 3),
alert_message TEXT,
resolved BOOLEAN DEFAULT FALSE,
resolved_at TIMESTAMPTZ,
created_at TIMESTAMPTZ DEFAULT NOW()
);
-- View per genealogia lotto (query ottimizzata)
CREATE OR REPLACE VIEW lot_genealogy AS
SELECT
fl.lot_id,
fl.product_name,
fl.product_gtin,
p.name AS producer_name,
p.country AS producer_country,
fl.production_date,
fl.expiry_date,
fl.lot_status,
fl.certifications,
fl.raw_materials,
fl.blockchain_tx,
json_agg(
json_build_object(
'event_id', te.event_id,
'event_type', te.event_type,
'actor_name', te.actor_name,
'location', te.location_name,
'timestamp', te.created_at,
'temperature', te.temperature_celsius,
'humidity', te.humidity_percent,
'blockchain', te.blockchain_tx
) ORDER BY te.created_at
) AS events
FROM food_lots fl
LEFT JOIN producers p ON fl.producer_gln = p.gln
LEFT JOIN trace_events te ON fl.lot_id = te.lot_id
GROUP BY fl.lot_id, fl.product_name, fl.product_gtin,
p.name, p.country, fl.production_date, fl.expiry_date,
fl.lot_status, fl.certifications, fl.raw_materials, fl.blockchain_tx;Průvodce náklady: Implementace pro potravinářské MSP
Nejčastější otázka italských potravinářských malých a středních podniků zní: „kolik stojí implementace digitální systém sledovatelnosti?" Odpověď silně závisí na měřítku, složitostí dodavatelského řetězce a úrovní technologických ambicí. Tady je jeden realistické vedení založené na projektech realizovaných v letech 2024–2025.
Náklady na implementaci Sledovatelnost podle úrovně splatnosti
| Úroveň | Scénář | Technologie | Nastavení Capex | Provozní doba/rok | Časová osa |
|---|---|---|---|---|---|
| Základní | Malý výrobce DOP, 1 továrna, 10-50 šarží/měsíc | QR Code GS1, software pro sledování SaaS, čtečka čárových kódů | 3 000-8 000 EUR | 1 200-3 600 EUR | 1-2 měsíce |
| Střední | Družstvo, 5-20 továren, 500-2 000 šarží/měsíc | HF/UHF RFID + QR GS1 Digital Link + teplota IoT + EPCIS API | 40 000-120 000 EUR | 8 000-24 000 EUR | 4-8 měsíců |
| Moderní | DOP Consortium, 50+ výrobců, kompletní dodavatelský řetězec | Vše výše uvedené + Blockchain Hyperledger Fabric, IoT studeného řetězce, spotřebitelská aplikace | 200 000-800 000 EUR | 50 000-150 000 EUR | 12-24 měsíců |
| Podnik | Rozsáhlý maloobchod, globální dodavatelský řetězec s mnoha zainteresovanými stranami | IBM Food Trust / vlastní platforma, plné pokrytí RFID, analytika AI | 1M-5M EUR | 200 000-500 000 EUR | 18-36 měsíců |
RFID vs NFC vs QR kód: Kdy použít co
| Kritérium | QR kód GS1 | NFC | UHF RFID | HF RFID |
|---|---|---|---|---|
| Cena za jednotku | Minimum (pouze tisk) | Nízká (0,20–2 EUR) | Minimum (0,05–0,30 EUR) | Střední (0,30–1,50 EUR) |
| Rychlost čtení | 1 najednou | 1 najednou | 1000+ současně | 10-100 současně |
| Tváří v tvář spotřebiteli | Optimální | Optimální | Nevhodné | Omezený |
| Automatizovaná logistika | Nevhodné (přímá viditelnost) | Nevhodné | Vynikající | Dobrý |
| Odolnost vůči vlhkému prostředí | Špatný (mokrý papír) | Dobrý | Špatný (efekt vody) | Dobrý |
| Boj proti padělání | Nízká (klonovatelná) | Střední (podepsáno NDEF) | Nízký | Vysoká (p-Chip, Crypto tag) |
| Doporučeno pro | Consumer Packaging, DL | Prémiové produkty, DOP | Palety, sklad, logistika | Jednotlivé produkty, léky, DOP |
Pobídky a financování: Jak mohou potravinářské MSP získat přístup k PNRR a přechodu 5.0
Implementace pokročilých systémů digitální sledovatelnosti spadá do několika kategorií pobídková opatření dostupná pro italské potravinářské společnosti v letech 2025–2026. Znalost dostupných nástrojů může výrazně snížit čisté náklady investice.
Hlavní motivační nástroje pro digitální sledovatelnost potravin
- Přechod 5.0 (PNRR): Daňová sleva až 45 %. investice do digitálu a udržitelnosti výslovně zahrnují systémy internetu věcí, senzory a software pro sledovatelnost. Celkový rozpočet 6,3 miliardy EUR a přístupné přes GSE. Pozor: je nutné propojení se systémem vedení společnosti a podrobná technická dokumentace.
- Chytrý agropotravinářský tendr (MISE/MITE): Nenávratné financování (30-50 % způsobilých výdajů) na projekty digitalizace v dodavatelských řetězcích zemědělsko-potravinářský. Zahrnuje RFID hardware, IoT a vývoj softwaru na zakázku.
- Regionální PSR (Plán rozvoje venkova): Podporuje velikost 4.2 včetně investic do zpracování a uvádění zemědělských produktů na trh digitalizace sledovatelnosti. Obvykle spolufinancování EU + region 40–65 % způsobilých výdajů pro mikropodniky a malé a střední podniky.
- Horizon Europe (skupina 6): Pro výzkumná a inovační konsorcia, Evropské financování až do výše 100 % na výzkumné projekty v oblasti sledovatelnosti potravin s univerzitními partnery nebo výzkumnými centry.
- Sabatini zelená: Zvýhodněné financování investic do digitální technologie (včetně RFID čteček, IoT bran, serverů) s přispěním s úrokem 3,575 % při maximální výši úvěru 4 miliony EUR.
Plán implementace: 12 měsíců pro podnikový systém
Plán implementace sledovatelnosti potravin – postupný přístup
| Fáze | Období | Aktivita | Výstupy |
|---|---|---|---|
| Fáze 0: Hodnocení | měsíc 1 | Mapování dodavatelského řetězce, aplikovatelná regulační analýza, identifikace CTE, analýza mezer | Obchodní případ, architektonický plán |
| Fáze 1: Data Foundation | Měsíce 2-3 | Nastavení PostgreSQL, schéma databáze, základní API, firemní prefix GS1, přiřazení GTIN | Provozní databáze, API v1 |
| Fáze 2: Identifikace | Měsíce 3-4 | Nasazení klíčových bodů čtečky RFID, tisk QR kódu GS1 DL, komplexní testování | Systémové ID fungující na 1 lince |
| Fáze 3: IoT Cold Chain | Měsíce 4-6 | Nasaďte teplotní senzory, bránu MQTT, řídicí panel Grafana, výstrahu HACCP | Nepřetržité monitorování chladícího řetězce |
| Fáze 4: Blockchain | Měsíce 6-9 | Nastavení Hyperledger Fabric (nebo SaaS), začlenění partnera do dodavatelského řetězce, nasazení řetězového kódu | Provozní vrstva důvěryhodnosti blockchainu |
| Fáze 5: Spotřebitel | Měsíce 9-11 | Spotřebitelská webová aplikace (QR scan), vícejazyčná lokalizace, analytika využití | Živý spotřebitelský portál |
| Fáze 6: Schody | Měsíce 11-12 | Začlenění dalších dodavatelů, optimalizace výkonu, soulad s EPCIS 2.0 | Systém připravený na plnou výrobu |
Závěry: Sledovatelnost jako konkurenční výhoda
Digitální sledovatelnost potravin prošla fází testování a inovace: v roce 2025 je to konkurenční nutnost a pro mnoho kategorií produktů hrozící regulační povinnost. Globální trh, který předčí 23 miliard dolarů v roce 2025, směrnice FSMA 204 s prodloužením souladu do roku 2028 (ale nevyhnutelné) a Evropský digitální produktový pas nastiňují budoucnost, ve které každý potravinářský výrobek bude mít ověřitelnou digitální historii.
Technologie je vyspělá: UHF RFID za 5 eurocentů za štítek, digitální QR kód GS1 Odkazy přístupné komukoli, open-source Hyperledger Fabric s povolením blockchainu, BLE senzory pro chladící řetězec za 20-30 eur za kus. Skutečná výzva není technologická: a správa (kdo koordinuje dodavatelský řetězec?), onboarding (jak zapojit malé a střední podniky). méně digitalizované?) a obchodní modely (kdo platí, kdo má prospěch?).
Pro italské malé a střední podniky je případ Parmigiano Reggiano replikovatelným modelem: začněte s konsorciem, které koordinuje implementaci, má přístup k pobídkám PNRR a PSR, zvolit inkrementální přístup, který přináší hodnotu již ve fázi 1 (identifikace lotto digital) a postupem času přidává vrstvy sofistikovanosti. ROI není okamžitá, ale kombinace prémiových cen, stažení a snížení nákladů na přístup na nové trhy činí investici ekonomicky udržitelnou do 3-5 let.
Užitečné odkazy a zdroje
- GS1 Itálie: gs1it.org - Registrace prefixu společnosti, standard EPCIS 2.0
- Tkanina Hyperledger: hyperledger-fabric.readthedocs.io - Oficiální dokumentace
- Zdroje FDA FSMA 204: FDA.gov
- Kvalita – blockchain CHOP/CHZO: kvalita.it
- Strategie z farmy na vidličku: food.ec.europa.eu
Další článek ze série FoodTech
V dalším článku ze série se podíváme na to Počítačové vidění pro ovládání Kvalita jídla s PyTorch a YOLO: jak zavést kontrolní systémy automatické vidění pro detekci vad potravinářských výrobků na výrobní lince, s architekturou v reálném čase, školením a zaváděcím potrubím na průmyslovém hardwaru.
Pokračujte ve sledování série FoodTech na federicocalo.dev pro všechna budoucí vydání technické poznatky.