File size: 5,353 Bytes
83d7647 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | import ast
import re
from collections.abc import Iterable, Mapping
import torch
import torch.nn as nn
from huggingface_hub import PyTorchModelHubMixin
class DataProcessor:
def __init__(self):
self.vocab = None
def _check_readiness(self) -> bool:
if self.vocab is None:
raise ValueError('Missing vocab. Ensure loading respective vocab.json.')
@staticmethod
def _parse_tags(tags) -> list:
"""Read tag lists, including list strings loaded from CSV."""
if isinstance(tags, Mapping):
return []
if isinstance(tags, str):
control_escapes = {code: f'\\x{code:02x}' for code in range(32)}
values = []
for match in re.finditer('\'(?:\\\\.|[^\'\\\\])*\'|"(?:\\\\.|[^"\\\\])*"', tags):
try:
values.append(ast.literal_eval(match.group(0).translate(control_escapes)))
except (SyntaxError, ValueError):
continue
elif isinstance(tags, Iterable):
values = tags
else:
return []
return values
@staticmethod
def preprocess_tags(tags) -> list[str]:
"""Normalize tags using the vocabulary extraction rules."""
cleaned = []
for value in DataProcessor._parse_tags(tags):
if not isinstance(value, str):
continue
tag = value.lower()
for old, new in ((' - ', ''), ('-', ''), ('.', ''), ('|', ''), ('ä', 'a'), ('ö', 'o'), ('ü', 'u')):
tag = tag.replace(old, new)
if tag:
cleaned.append(tag)
return cleaned
def normalize_tags(self, tags) -> list[str]:
"""Preprocess raw or normalized tags and keep only known tags."""
self._check_readiness()
return [tag for tag in self.preprocess_tags(tags) if tag in self.known_tags_dict]
def vectorize_tags(self, tags) -> torch.Tensor:
"""Creates onehot-encoded vector of used tags for classification.
Ignores unknown tags.
Returns vector full of zeros if no valid tags were found or tags to normalize were [] or `None`"""
self._check_readiness()
base = torch.zeros(self.nr_of_known_tags, dtype=torch.float32)
for p_tag in self.normalize_tags(tags):
base[int(self.known_tags_dict[p_tag])] = 1
return base
def process_api(self, response_dict: dict) -> torch.Tensor:
"""Vectorize an OFF product dictionary; missing tags produce a zero vector."""
self._check_readiness()
return self.vectorize_tags(response_dict.get('categories_tags'))
class MLP(nn.Module, PyTorchModelHubMixin):
def __init__(self, input_features:int):
super().__init__()
self.class_mapping = {
0: "Obst & Gemüse",
1: "Fleisch, Fisch & Alternativen",
2: "Milchprodukte & Eier",
3: "Brot & Backwaren",
4: "Nudeln, Reis & Getreide",
5: "Fertigprodukte",
6: "Backzutaten",
7: "Getränke",
8: "Snacks & Süßigkeiten",
9: "Öle, Gewürze & Saucen",
10: "Brotaufstriche",
11: "Sonstiges"
}
self.input_layer = nn.Linear(in_features=input_features, out_features=2000) # classes 0-11
self.hidden1 = nn.Linear(self.input_layer.out_features, 1000)
self.hidden2 = nn.Linear(self.hidden1.out_features, 100)
self.hidden3 = nn.Linear(self.hidden2.out_features, 12)
self.softmax = nn.Softmax(1)
self.leaky_r = nn.LeakyReLU(inplace=True)
self.dropout1 = nn.Dropout(p=.2)
self.dropout2 = nn.Dropout(p=.2)
self.forward_seq = nn.Sequential(
self.input_layer,
self.leaky_r,
self.hidden1,
self.dropout1,
self.leaky_r,
self.hidden2,
self.dropout2,
self.leaky_r,
self.hidden3
)
def forward(self, X:torch.Tensor) -> torch.Tensor:
"""Performs forward pass."""
return self.forward_seq(X)
def predict(self, X, string_classes:bool=True) -> torch.Tensor:
if X.ndim == 1:
X = X.unsqueeze(0)
preds = self.forward_seq(X)
res = int(torch.argmax(preds, dim=1).cpu())
return self.class_mapping[res] if string_classes else res
class FoodClassifier(MLP):
"""MLP with preprocessing and the standard Hugging Face Hub mixin."""
def __init__(self, vocab: list[str]):
super().__init__(input_features=len(vocab))
# The mixin saves the ordered vocabulary in config.json.
self.processor = DataProcessor()
self.processor.vocab = list(vocab)
self.processor.known_tags_dict = {tag: idx for idx, tag in enumerate(vocab)}
self.processor.nr_of_known_tags = len(vocab)
@torch.inference_mode()
def predict(self, product: dict) -> str:
"""Return the category name for a product containing categories_tags."""
vector = self.processor.process_api(product).to(next(self.parameters()).device)
return super().predict(vector, string_classes=True)
|