Processor
TransformersQAProcessor¶
Question Answering Data Processor. Use this class to generate tensor inputs from human legible text/string data. This class can be used with a majority of the Bert architecture transformer models with the span-based extractive, Question Answering predictive head from Hugging Face.
Usage:
>>> processor = TRansformersQAProcessor(model_name_or_path="distilbert-base-cased-distilled-squad")
>>> processor.process(query=["string"], context["string"])
**Parameters:**
* **model_name_or_path** - String defining HF question answering model/tokenizer's name
Source code in fastnn/processors/nlp/question_answering.py
class TransformersQAProcessor(Processor):
"""Question Answering Data Processor. Use this class to generate tensor inputs from human legible text/string data.
This class can be used with a majority of the Bert architecture transformer models with the span-based extractive,
Question Answering predictive head from Hugging Face.
Usage:
```python
>>> processor = TRansformersQAProcessor(model_name_or_path="distilbert-base-cased-distilled-squad")
>>> processor.process(query=["string"], context["string"])
**Parameters:**
* **model_name_or_path** - String defining HF question answering model/tokenizer's name
```
"""
def __init__(
self, model_name_or_path: str = "distilbert-base-cased-distilled-squad"
):
self.tokenizer = AutoTokenizer.from_pretrained(
model_name_or_path, use_fast=False
) # Can't use fast tokenizer yet for QA `use_fast=True`
def process(
self,
query: List[str],
context: List[str],
max_seq_length: int = 512,
doc_stride: int = 128,
max_query_length: int = 64,
) -> Tuple[List[SquadExample], List[SquadFeatures], Dataset]:
"""Generate torch `Dataset` object from query/context string pairs using specified tokenizer from HF.
This provides clear tensor input representations for compatible models.
Returns a tuple `Dataset` and matching `SquadFeatures`
* **query** - List of query strings, must be same length as `context`
* **context** - List of context strings, must be same length as `query`
* **max_seq_length** - Maximum context token length. Check model configs to see max sequence length the model was trained with
* **doc_stride** - Number of token strides to take when splitting up context into chunks of size `max_seq_length`
* **max_query_length** - Maximum token length for queries
"""
examples = self._generate_squad_examples(query=query, context=context)
features, dataset = squad_convert_examples_to_features(
examples,
self.tokenizer,
max_seq_length=max_seq_length,
doc_stride=doc_stride,
max_query_length=max_query_length,
is_training=False,
return_dataset="pt",
threads=1,
)
return examples, features, dataset
def process_batch(
self,
query: List,
context: List,
mini_batch_size: int = 8,
max_seq_length: int = 512,
doc_stride: int = 128,
max_query_length: int = 64,
use_gpu: bool = False,
) -> Tuple[List[SquadExample], List[SquadFeatures], DataLoader]:
"""Generate torch `DataLoader` object from query/context string pairs using specified tokenizer from HF.
This provides clear tensor input representations for compatible models in an easy to use batch
Returns a tuple of (List[`SquadExample`], List[`SquadFeatures`], `DataLoader`)
* **query** - List of query strings, must be same length as `context`
* **context** - List of context strings, must be same length as `query`
* **mini_batch_size** - Batch size for inference
* **max_seq_length** - Maximum context token length. Check model configs to see max sequence length the model was trained with
* **doc_stride** - Number of token strides to take when splitting up context into chunks of size `max_seq_length`
* **max_query_length** - Maximum token length for queries
* **use_gpu** - Bool for using gpu or cpu. If set True but no gpu devices available, model will default to using cpu
"""
if use_gpu:
if torch.cuda.is_available():
device = torch.device("cuda")
else:
logger.info("GPU not available")
device = torch.device("cpu")
else:
device = torch.device("cpu")
examples, features, dataset = self.process(
query=query,
context=context,
max_seq_length=max_seq_length,
doc_stride=doc_stride,
max_query_length=max_query_length,
)
dataloader = DataLoader(
dataset,
batch_size=mini_batch_size,
collate_fn=lambda x: [t.to(device) for t in self._qa_collate_fn(x)],
)
return examples, features, dataloader
def process_output(
self,
outputs: List[Tuple[torch.Tensor, torch.Tensor, torch.Tensor]],
examples: List[SquadExample],
features: List[SquadFeatures],
n_best_size: int = 5,
max_answer_length: int = 10,
do_lower_case: bool = False,
verbose_logging: bool = False,
version_2_with_negative: bool = False,
null_score_diff_threshold: float = 0.0,
):
pass
def process_output_batch(
self,
outputs: List[Tuple[torch.Tensor, torch.Tensor, torch.Tensor]],
examples: List[SquadExample],
features: List[SquadFeatures],
n_best_size: int = 5,
max_answer_length: int = 64,
do_lower_case: bool = False,
verbose_logging: bool = False,
version_2_with_negative: bool = False,
null_score_diff_threshold: float = 0.0,
) -> Tuple[OrderedDict, OrderedDict]:
"""Process output of Transformers QA model into human legible results.
* **outputs** - List of batch output tensors from a model's forward pass
* **examples** - List of `SquadExample` objects for each original context/query pair used as input. This is returned from the built-in `process()` or `process_batch()` methods
* **features** - List of `SquadFeature` objects for each context/query pair over the original doc_stride lengths. This is also returned from the built-in `process()` or `process_batch()` methods
* **n_best_size** - Number of top n results you want
* **max_answer_length** - Maximum token length for answers that are returned
* **do_lower_case** - Set as `True` if using uncased QA models
* **verbose_logging** - Set True if you want prediction verbose loggings
* **version_2_with_negative** - Set as True if using QA model with SQUAD2.0
* **null_score_diff_threshold** - Threshold for predicting null(no answer) in Squad 2.0 Model. Default is 0.0. Raise this if you want fewer null answers
"""
# Generate results per example query
all_results: List[SquadResult] = []
for output in outputs:
example_indices = output[2]
for i, example_index in enumerate(example_indices):
start_logits = self._to_list(output[0][i])
end_logits = self._to_list(output[1][i])
eval_feature = features[example_index[0].item()]
unique_id = int(eval_feature.unique_id)
result = SquadResult(unique_id, start_logits, end_logits)
all_results.append(result)
# Compute predictions based off logits on a per example basis
answers, n_best = compute_predictions_logits(
all_examples=examples,
all_features=features,
all_results=all_results,
n_best_size=n_best_size,
max_answer_length=max_answer_length,
do_lower_case=do_lower_case,
verbose_logging=verbose_logging,
version_2_with_negative=version_2_with_negative,
null_score_diff_threshold=null_score_diff_threshold,
tokenizer=self.tokenizer,
)
return answers, n_best
def _qa_collate_fn(self, data):
batch = default_collate(data)
# Generate same batch dims for scalars to address future batch inferencing
batch[3].unsqueeze_(1)
batch[4].unsqueeze_(1)
return batch
def _to_list(self, tensor: torch.Tensor):
return tensor.detach().cpu().tolist()
def _generate_squad_examples(
self, query: List[str], context: List[str]
) -> List[SquadExample]:
"""Generate HF Squad Example objects with query/context pairs"""
assert len(query) == len(context)
examples = []
title = "qa"
is_impossible = False
answer_text = None
start_position_character = None
answers = ["answer"]
for idx, (q, c) in enumerate(zip(query, context)):
example = SquadExample(
qas_id=str(idx),
question_text=q,
context_text=c,
answer_text=answer_text,
start_position_character=start_position_character,
title=title,
is_impossible=is_impossible,
answers=answers,
)
examples.append(example)
return examples
process(self, query, context, max_seq_length=512, doc_stride=128, max_query_length=64)
¶
Generate torch Dataset object from query/context string pairs using specified tokenizer from HF.
This provides clear tensor input representations for compatible models.
Returns a tuple Dataset and matching SquadFeatures
- query - List of query strings, must be same length as
context - context - List of context strings, must be same length as
query - max_seq_length - Maximum context token length. Check model configs to see max sequence length the model was trained with
- doc_stride - Number of token strides to take when splitting up context into chunks of size
max_seq_length - max_query_length - Maximum token length for queries
Source code in fastnn/processors/nlp/question_answering.py
def process(
self,
query: List[str],
context: List[str],
max_seq_length: int = 512,
doc_stride: int = 128,
max_query_length: int = 64,
) -> Tuple[List[SquadExample], List[SquadFeatures], Dataset]:
"""Generate torch `Dataset` object from query/context string pairs using specified tokenizer from HF.
This provides clear tensor input representations for compatible models.
Returns a tuple `Dataset` and matching `SquadFeatures`
* **query** - List of query strings, must be same length as `context`
* **context** - List of context strings, must be same length as `query`
* **max_seq_length** - Maximum context token length. Check model configs to see max sequence length the model was trained with
* **doc_stride** - Number of token strides to take when splitting up context into chunks of size `max_seq_length`
* **max_query_length** - Maximum token length for queries
"""
examples = self._generate_squad_examples(query=query, context=context)
features, dataset = squad_convert_examples_to_features(
examples,
self.tokenizer,
max_seq_length=max_seq_length,
doc_stride=doc_stride,
max_query_length=max_query_length,
is_training=False,
return_dataset="pt",
threads=1,
)
return examples, features, dataset
process_batch(self, query, context, mini_batch_size=8, max_seq_length=512, doc_stride=128, max_query_length=64, use_gpu=False)
¶
Generate torch DataLoader object from query/context string pairs using specified tokenizer from HF.
This provides clear tensor input representations for compatible models in an easy to use batch
Returns a tuple of (List[SquadExample], List[SquadFeatures], DataLoader)
- query - List of query strings, must be same length as
context - context - List of context strings, must be same length as
query - mini_batch_size - Batch size for inference
- max_seq_length - Maximum context token length. Check model configs to see max sequence length the model was trained with
- doc_stride - Number of token strides to take when splitting up context into chunks of size
max_seq_length - max_query_length - Maximum token length for queries
- use_gpu - Bool for using gpu or cpu. If set True but no gpu devices available, model will default to using cpu
Source code in fastnn/processors/nlp/question_answering.py
def process_batch(
self,
query: List,
context: List,
mini_batch_size: int = 8,
max_seq_length: int = 512,
doc_stride: int = 128,
max_query_length: int = 64,
use_gpu: bool = False,
) -> Tuple[List[SquadExample], List[SquadFeatures], DataLoader]:
"""Generate torch `DataLoader` object from query/context string pairs using specified tokenizer from HF.
This provides clear tensor input representations for compatible models in an easy to use batch
Returns a tuple of (List[`SquadExample`], List[`SquadFeatures`], `DataLoader`)
* **query** - List of query strings, must be same length as `context`
* **context** - List of context strings, must be same length as `query`
* **mini_batch_size** - Batch size for inference
* **max_seq_length** - Maximum context token length. Check model configs to see max sequence length the model was trained with
* **doc_stride** - Number of token strides to take when splitting up context into chunks of size `max_seq_length`
* **max_query_length** - Maximum token length for queries
* **use_gpu** - Bool for using gpu or cpu. If set True but no gpu devices available, model will default to using cpu
"""
if use_gpu:
if torch.cuda.is_available():
device = torch.device("cuda")
else:
logger.info("GPU not available")
device = torch.device("cpu")
else:
device = torch.device("cpu")
examples, features, dataset = self.process(
query=query,
context=context,
max_seq_length=max_seq_length,
doc_stride=doc_stride,
max_query_length=max_query_length,
)
dataloader = DataLoader(
dataset,
batch_size=mini_batch_size,
collate_fn=lambda x: [t.to(device) for t in self._qa_collate_fn(x)],
)
return examples, features, dataloader
process_output_batch(self, outputs, examples, features, n_best_size=5, max_answer_length=64, do_lower_case=False, verbose_logging=False, version_2_with_negative=False, null_score_diff_threshold=0.0)
¶
Process output of Transformers QA model into human legible results.
- outputs - List of batch output tensors from a model's forward pass
- examples - List of
SquadExampleobjects for each original context/query pair used as input. This is returned from the built-inprocess()orprocess_batch()methods - features - List of
SquadFeatureobjects for each context/query pair over the original doc_stride lengths. This is also returned from the built-inprocess()orprocess_batch()methods - n_best_size - Number of top n results you want
- max_answer_length - Maximum token length for answers that are returned
- do_lower_case - Set as
Trueif using uncased QA models - verbose_logging - Set True if you want prediction verbose loggings
- version_2_with_negative - Set as True if using QA model with SQUAD2.0
- null_score_diff_threshold - Threshold for predicting null(no answer) in Squad 2.0 Model. Default is 0.0. Raise this if you want fewer null answers
Source code in fastnn/processors/nlp/question_answering.py
def process_output_batch(
self,
outputs: List[Tuple[torch.Tensor, torch.Tensor, torch.Tensor]],
examples: List[SquadExample],
features: List[SquadFeatures],
n_best_size: int = 5,
max_answer_length: int = 64,
do_lower_case: bool = False,
verbose_logging: bool = False,
version_2_with_negative: bool = False,
null_score_diff_threshold: float = 0.0,
) -> Tuple[OrderedDict, OrderedDict]:
"""Process output of Transformers QA model into human legible results.
* **outputs** - List of batch output tensors from a model's forward pass
* **examples** - List of `SquadExample` objects for each original context/query pair used as input. This is returned from the built-in `process()` or `process_batch()` methods
* **features** - List of `SquadFeature` objects for each context/query pair over the original doc_stride lengths. This is also returned from the built-in `process()` or `process_batch()` methods
* **n_best_size** - Number of top n results you want
* **max_answer_length** - Maximum token length for answers that are returned
* **do_lower_case** - Set as `True` if using uncased QA models
* **verbose_logging** - Set True if you want prediction verbose loggings
* **version_2_with_negative** - Set as True if using QA model with SQUAD2.0
* **null_score_diff_threshold** - Threshold for predicting null(no answer) in Squad 2.0 Model. Default is 0.0. Raise this if you want fewer null answers
"""
# Generate results per example query
all_results: List[SquadResult] = []
for output in outputs:
example_indices = output[2]
for i, example_index in enumerate(example_indices):
start_logits = self._to_list(output[0][i])
end_logits = self._to_list(output[1][i])
eval_feature = features[example_index[0].item()]
unique_id = int(eval_feature.unique_id)
result = SquadResult(unique_id, start_logits, end_logits)
all_results.append(result)
# Compute predictions based off logits on a per example basis
answers, n_best = compute_predictions_logits(
all_examples=examples,
all_features=features,
all_results=all_results,
n_best_size=n_best_size,
max_answer_length=max_answer_length,
do_lower_case=do_lower_case,
verbose_logging=verbose_logging,
version_2_with_negative=version_2_with_negative,
null_score_diff_threshold=null_score_diff_threshold,
tokenizer=self.tokenizer,
)
return answers, n_best
TransformersTokenTaggingProcessor¶
Token Tagging Data Processor. Use this class to generate tensor inputs from human legible text/string data. This class can be used with a majority of the Bert architecture transformer models with a token-level predictive head for token classification from Hugging Face.
Usage:
>>> processor = TransformersTokenTaggingProcessor(model_name_or_path="dbmdz/bert-large-cased-finetuned-conll03-english")
>>> processor.process(text=["string"])
**Parameters:**
* **model_name_or_path** - String defining HF token tagging model/tokenizer's name
* **label_strings** - List of strings that specify label strings with index as key for this specific processor
Source code in fastnn/processors/nlp/token_tagging.py
class TransformersTokenTaggingProcessor(Processor):
"""Token Tagging Data Processor. Use this class to generate tensor inputs from human legible text/string data.
This class can be used with a majority of the Bert architecture transformer models with a token-level predictive head
for token classification from Hugging Face.
Usage:
```python
>>> processor = TransformersTokenTaggingProcessor(model_name_or_path="dbmdz/bert-large-cased-finetuned-conll03-english")
>>> processor.process(text=["string"])
**Parameters:**
* **model_name_or_path** - String defining HF token tagging model/tokenizer's name
* **label_strings** - List of strings that specify label strings with index as key for this specific processor
```
"""
def __init__(
self,
model_name_or_path: str = "dbmdz/bert-large-cased-finetuned-conll03-english",
label_strings: List[str] = [
"O", # Outside of a named entity
"B-MISC", # Beginning of a miscellaneous entity right after another miscellaneous entity
"I-MISC", # Miscellaneous entity
"B-PER", # Beginning of a person's name right after another person's name
"I-PER", # Person's name
"B-ORG", # Beginning of an organisation right after another organisation
"I-ORG", # Organisation
"B-LOC", # Beginning of a location right after another location
"I-LOC", # Location
],
):
self.tokenizer = AutoTokenizer.from_pretrained(
model_name_or_path, use_fast=True
)
self.label_strings = label_strings
def process(
self,
text: List[str],
max_seq_length: int = 512,
) -> Tuple:
"""Generate torch `Dataset` object from query/context string pairs using specified tokenizer from HF.
This provides clear tensor input representations for compatible models.
Returns a `Dataset`
* **text** - List of text strings
* **max_seq_length** - Maximum context token length. Check model configs to see max sequence length the model was trained with
"""
tokens = self.tokenizer(
text=text, return_tensors="pt", padding="max_length", truncation=True
)
dataset = TensorDataset(
tokens["input_ids"],
tokens["attention_mask"],
)
return dataset
def process_batch(
self,
text: List,
mini_batch_size: int = 8,
max_seq_length: int = 512,
use_gpu: bool = False,
) -> DataLoader:
"""Generate torch `DataLoader` object from text strings using specified tokenizer from HF.
This provides clear tensor input representations for compatible models in an easy to use batch
Returns a `DataLoader`
* **text** - List of text strings
* **mini_batch_size** - Batch size for inference
* **max_seq_length** - Maximum context token length. Check model configs to see max sequence length the model was trained with
* **use_gpu** - Bool for using gpu or cpu. If set True but no gpu devices available, model will default to using cpu
"""
if use_gpu:
if torch.cuda.is_available():
device = torch.device("cuda")
else:
logger.info("GPU not available")
device = torch.device("cpu")
else:
device = torch.device("cpu")
dataset = self.process(text=text)
dataloader = DataLoader(
dataset,
batch_size=mini_batch_size,
collate_fn=lambda x: [t.to(device) for t in self._collate_fn(x)],
)
return dataloader
def process_output(
self,
outputs: List[Tuple[torch.Tensor, torch.Tensor, torch.Tensor]],
):
pass
def process_output_batch(self, outputs: List) -> List[List[Tuple[str, str]]]:
"""Process output of Transformers NER model
* **outputs** - List of batch output tensors from a model's forward pass
"""
results = []
for logits, input_ids in outputs:
tokens_batch = [self.tokenizer.convert_ids_to_tokens(i) for i in input_ids]
argmax_batch = [torch.argmax(o, dim=1) for o in logits]
for i in range(len(tokens_batch)):
# Filter out padding
results.append(
[
(token, self.label_strings[prediction])
for token, prediction in zip(
tokens_batch[i], argmax_batch[i].cpu().numpy()
)
if token != "[PAD]"
]
)
return results
def _collate_fn(self, data):
batch = default_collate(data)
return batch
process(self, text, max_seq_length=512)
¶
Generate torch Dataset object from query/context string pairs using specified tokenizer from HF.
This provides clear tensor input representations for compatible models.
Returns a Dataset
- text - List of text strings
- max_seq_length - Maximum context token length. Check model configs to see max sequence length the model was trained with
Source code in fastnn/processors/nlp/token_tagging.py
def process(
self,
text: List[str],
max_seq_length: int = 512,
) -> Tuple:
"""Generate torch `Dataset` object from query/context string pairs using specified tokenizer from HF.
This provides clear tensor input representations for compatible models.
Returns a `Dataset`
* **text** - List of text strings
* **max_seq_length** - Maximum context token length. Check model configs to see max sequence length the model was trained with
"""
tokens = self.tokenizer(
text=text, return_tensors="pt", padding="max_length", truncation=True
)
dataset = TensorDataset(
tokens["input_ids"],
tokens["attention_mask"],
)
return dataset
process_batch(self, text, mini_batch_size=8, max_seq_length=512, use_gpu=False)
¶
Generate torch DataLoader object from text strings using specified tokenizer from HF.
This provides clear tensor input representations for compatible models in an easy to use batch
Returns a DataLoader
- text - List of text strings
- mini_batch_size - Batch size for inference
- max_seq_length - Maximum context token length. Check model configs to see max sequence length the model was trained with
- use_gpu - Bool for using gpu or cpu. If set True but no gpu devices available, model will default to using cpu
Source code in fastnn/processors/nlp/token_tagging.py
def process_batch(
self,
text: List,
mini_batch_size: int = 8,
max_seq_length: int = 512,
use_gpu: bool = False,
) -> DataLoader:
"""Generate torch `DataLoader` object from text strings using specified tokenizer from HF.
This provides clear tensor input representations for compatible models in an easy to use batch
Returns a `DataLoader`
* **text** - List of text strings
* **mini_batch_size** - Batch size for inference
* **max_seq_length** - Maximum context token length. Check model configs to see max sequence length the model was trained with
* **use_gpu** - Bool for using gpu or cpu. If set True but no gpu devices available, model will default to using cpu
"""
if use_gpu:
if torch.cuda.is_available():
device = torch.device("cuda")
else:
logger.info("GPU not available")
device = torch.device("cpu")
else:
device = torch.device("cpu")
dataset = self.process(text=text)
dataloader = DataLoader(
dataset,
batch_size=mini_batch_size,
collate_fn=lambda x: [t.to(device) for t in self._collate_fn(x)],
)
return dataloader
process_output_batch(self, outputs)
¶
Process output of Transformers NER model
- outputs - List of batch output tensors from a model's forward pass
Source code in fastnn/processors/nlp/token_tagging.py
def process_output_batch(self, outputs: List) -> List[List[Tuple[str, str]]]:
"""Process output of Transformers NER model
* **outputs** - List of batch output tensors from a model's forward pass
"""
results = []
for logits, input_ids in outputs:
tokens_batch = [self.tokenizer.convert_ids_to_tokens(i) for i in input_ids]
argmax_batch = [torch.argmax(o, dim=1) for o in logits]
for i in range(len(tokens_batch)):
# Filter out padding
results.append(
[
(token, self.label_strings[prediction])
for token, prediction in zip(
tokens_batch[i], argmax_batch[i].cpu().numpy()
)
if token != "[PAD]"
]
)
return results
ObjectDetectionProcessor¶
Object Detection processor dealing with image files or 3xHxW formatted images and boxes, scores, labels out processing. Since most resizing and padding transforms are done by the object detection models in PyTorch, datasets and dataloaders willl generate batches of images as lists.
Usage:
>>> processor = ObjectDetectionProcessor()
>>> processor.process(file_paths=["file_path.png"])
**Parameters:**
* **label_strings** - List of strings that specify label strings with index as key for this specific processor
Source code in fastnn/processors/cv/object_detection.py
class ObjectDetectionProcessor(Processor):
"""Object Detection processor dealing with image files or 3xHxW formatted images and boxes, scores, labels out processing.
Since most resizing and padding transforms are done by the object detection models in PyTorch, datasets and dataloaders willl
generate batches of images as lists.
Usage:
```python
>>> processor = ObjectDetectionProcessor()
>>> processor.process(file_paths=["file_path.png"])
**Parameters:**
* **label_strings** - List of strings that specify label strings with index as key for this specific processor
```
"""
def __init__(self, label_strings: List[str]):
self.label_strings = label_strings
def process(
self,
dir_path: str,
transforms: Optional[Callable] = ConvertImageDtype(torch.float),
) -> Dataset:
"""Generate torch `Dataset` object from list of file paths or image Tensors.
This provides clear tensor input representations for compatible models.
Returns a Dataset
* **dir_path** - String path to directory of images you'd like to process
"""
dataset = ImageDataset(root=dir_path, transforms=transforms)
return dataset
def process_batch(
self,
dir_path: str,
transforms: Optional[Callable] = ConvertImageDtype(torch.float),
mini_batch_size: int = 8,
use_gpu: bool = False,
) -> DataLoader:
"""Generate torch `Dataloader` object from data directory path.
This provides clear tensor input representations for compatible models.
Returns a `Dataloader`
* **dir_path** - String path to directory of images you'd like to process
* **mini_batch_size** - Batch size for inference
* **use_gpu** - Bool for using gpu or cpu. If set True but no gpu devices available, model will default to using cpu
"""
if use_gpu:
if torch.cuda.is_available():
device = torch.device("cuda")
else:
logger.info("GPU not available")
device = torch.device("cpu")
else:
device = torch.device("cpu")
dataset = self.process(dir_path=dir_path, transforms=transforms)
# Instead of a tensor batch, the lambda collate_fn will provide a list batch
dataloader = DataLoader(
dataset,
batch_size=mini_batch_size,
collate_fn=lambda x: [[t.to(device) for t in self._od_collate_fn(x)]],
)
return dataloader
def process_output(
self,
):
pass
def process_output_batch(
self, outputs: List[List[torch.Tensor]], dataset: Dataset
) -> List[List[Tuple[torch.Tensor, np.array]]]:
"""Process output of object detection model into human legible results.
Outputs from `FasterRCNNModule`
Returns batched results of list of list of tuples containing boxed images in tensor and numpy format
* **outputs** - List of batch output tensors from a model's forward pass
* **dataset** - Corresponding dataset with originial images matched with model outputs
"""
# Labeled Images
results = []
for idx, out in enumerate(outputs):
labeled_images = []
for label_idx in range(1, len(out), 3):
labels = [self.label_strings[o] for o in out[label_idx]]
unique_labels = set(labels)
label_colors_map = {}
for label in unique_labels:
label_colors_map[label] = tuple(
np.random.choice(range(256), size=3)
)
label_colors = [label_colors_map[label] for label in labels]
output_tensor, output_numpy = self.draw_bounding_boxes(
ConvertImageDtype(torch.uint8)(
dataset[idx * (len(out) // 3) + label_idx // 3]
),
out[label_idx - 1],
labels=labels,
colors=label_colors,
)
labeled_images.append((output_tensor, output_numpy))
results.append(labeled_images)
return results
def _od_collate_fn(self, data):
"""Custom collate fn to output dynamic image batches without same-dim requirements via. `stack`.
This is not technically a "correct" collate_fn for most of torch's vision models. Should be wrapped as a list
in the lambda collate fn.
"""
data = [img for img in data]
return data
@torch.no_grad()
def draw_bounding_boxes(
self,
image: torch.Tensor,
boxes: torch.Tensor,
labels: Optional[List[str]] = None,
colors: Optional[List[Union[str, Tuple[int, int, int]]]] = None,
width: int = 1,
font: Optional[str] = "arial.ttf",
font_size: int = 10,
) -> Tuple[torch.Tensor, np.array]:
"""
Added and modified from TorchVision utils.
Draws bounding boxes on given image.
The values of the input image should be uint8 between 0 and 255.
Args:
image (Tensor): Tensor of shape (C x H x W)
bboxes (Tensor): Tensor of size (N, 4) containing bounding boxes in (xmin, ymin, xmax, ymax) format. Note that
the boxes are absolute coordinates with respect to the image. In other words: `0 <= xmin < xmax < W` and
`0 <= ymin < ymax < H`.
labels (List[str]): List containing the labels of bounding boxes.
colors (List[Union[str, Tuple[int, int, int]]]): List containing the colors of bounding boxes. The colors can
be represented as `str` or `Tuple[int, int, int]`.
width (int): Width of bounding box.
font (str): A filename containing a TrueType font. If the file is not found in this filename, the loader may
also search in other directories, such as the `fonts/` directory on Windows or `/Library/Fonts/`,
`/System/Library/Fonts/` and `~/Library/Fonts/` on macOS.
font_size (int): The requested font size in points.
"""
if not isinstance(image, torch.Tensor):
raise TypeError(f"Tensor expected, got {type(image)}")
elif image.dtype != torch.uint8:
raise ValueError(f"Tensor uint8 expected, got {image.dtype}")
elif image.dim() != 3:
raise ValueError("Pass individual images, not batches")
ndarr = image.permute(1, 2, 0).numpy()
img_to_draw = Image.fromarray(ndarr)
img_boxes = boxes.to(torch.int64).tolist()
draw = ImageDraw.Draw(img_to_draw)
pixel_ratio = max(1, (max(ndarr.shape[0], ndarr.shape[1]) // 1000))
for i, bbox in enumerate(img_boxes):
color = None if colors is None else colors[i]
draw.rectangle(bbox, width=width * pixel_ratio, outline=color)
if labels is not None:
txt_font = (
ImageFont.load_default()
if font is None
else ImageFont.truetype(font=font, size=font_size * pixel_ratio)
)
draw.text((bbox[0], bbox[1]), labels[i], fill=color, font=txt_font)
return torch.from_numpy(np.array(img_to_draw)).permute(2, 0, 1), np.array(
img_to_draw
)
draw_bounding_boxes(self, image, boxes, labels=None, colors=None, width=1, font='arial.ttf', font_size=10)
¶
Added and modified from TorchVision utils. Draws bounding boxes on given image. The values of the input image should be uint8 between 0 and 255.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
image |
Tensor |
Tensor of shape (C x H x W) |
required |
bboxes |
Tensor |
Tensor of size (N, 4) containing bounding boxes in (xmin, ymin, xmax, ymax) format. Note that
the boxes are absolute coordinates with respect to the image. In other words: |
required |
labels |
List[str] |
List containing the labels of bounding boxes. |
None |
colors |
List[Union[str, Tuple[int, int, int]]] |
List containing the colors of bounding boxes. The colors can
be represented as |
None |
width |
int |
Width of bounding box. |
1 |
font |
str |
A filename containing a TrueType font. If the file is not found in this filename, the loader may
also search in other directories, such as the |
'arial.ttf' |
font_size |
int |
The requested font size in points. |
10 |
Source code in fastnn/processors/cv/object_detection.py
@torch.no_grad()
def draw_bounding_boxes(
self,
image: torch.Tensor,
boxes: torch.Tensor,
labels: Optional[List[str]] = None,
colors: Optional[List[Union[str, Tuple[int, int, int]]]] = None,
width: int = 1,
font: Optional[str] = "arial.ttf",
font_size: int = 10,
) -> Tuple[torch.Tensor, np.array]:
"""
Added and modified from TorchVision utils.
Draws bounding boxes on given image.
The values of the input image should be uint8 between 0 and 255.
Args:
image (Tensor): Tensor of shape (C x H x W)
bboxes (Tensor): Tensor of size (N, 4) containing bounding boxes in (xmin, ymin, xmax, ymax) format. Note that
the boxes are absolute coordinates with respect to the image. In other words: `0 <= xmin < xmax < W` and
`0 <= ymin < ymax < H`.
labels (List[str]): List containing the labels of bounding boxes.
colors (List[Union[str, Tuple[int, int, int]]]): List containing the colors of bounding boxes. The colors can
be represented as `str` or `Tuple[int, int, int]`.
width (int): Width of bounding box.
font (str): A filename containing a TrueType font. If the file is not found in this filename, the loader may
also search in other directories, such as the `fonts/` directory on Windows or `/Library/Fonts/`,
`/System/Library/Fonts/` and `~/Library/Fonts/` on macOS.
font_size (int): The requested font size in points.
"""
if not isinstance(image, torch.Tensor):
raise TypeError(f"Tensor expected, got {type(image)}")
elif image.dtype != torch.uint8:
raise ValueError(f"Tensor uint8 expected, got {image.dtype}")
elif image.dim() != 3:
raise ValueError("Pass individual images, not batches")
ndarr = image.permute(1, 2, 0).numpy()
img_to_draw = Image.fromarray(ndarr)
img_boxes = boxes.to(torch.int64).tolist()
draw = ImageDraw.Draw(img_to_draw)
pixel_ratio = max(1, (max(ndarr.shape[0], ndarr.shape[1]) // 1000))
for i, bbox in enumerate(img_boxes):
color = None if colors is None else colors[i]
draw.rectangle(bbox, width=width * pixel_ratio, outline=color)
if labels is not None:
txt_font = (
ImageFont.load_default()
if font is None
else ImageFont.truetype(font=font, size=font_size * pixel_ratio)
)
draw.text((bbox[0], bbox[1]), labels[i], fill=color, font=txt_font)
return torch.from_numpy(np.array(img_to_draw)).permute(2, 0, 1), np.array(
img_to_draw
)
process(self, dir_path, transforms=ConvertImageDtype())
¶
Generate torch Dataset object from list of file paths or image Tensors.
This provides clear tensor input representations for compatible models.
Returns a Dataset
- dir_path - String path to directory of images you'd like to process
Source code in fastnn/processors/cv/object_detection.py
def process(
self,
dir_path: str,
transforms: Optional[Callable] = ConvertImageDtype(torch.float),
) -> Dataset:
"""Generate torch `Dataset` object from list of file paths or image Tensors.
This provides clear tensor input representations for compatible models.
Returns a Dataset
* **dir_path** - String path to directory of images you'd like to process
"""
dataset = ImageDataset(root=dir_path, transforms=transforms)
return dataset
process_batch(self, dir_path, transforms=ConvertImageDtype(), mini_batch_size=8, use_gpu=False)
¶
Generate torch Dataloader object from data directory path.
This provides clear tensor input representations for compatible models.
Returns a Dataloader
- dir_path - String path to directory of images you'd like to process
- mini_batch_size - Batch size for inference
- use_gpu - Bool for using gpu or cpu. If set True but no gpu devices available, model will default to using cpu
Source code in fastnn/processors/cv/object_detection.py
def process_batch(
self,
dir_path: str,
transforms: Optional[Callable] = ConvertImageDtype(torch.float),
mini_batch_size: int = 8,
use_gpu: bool = False,
) -> DataLoader:
"""Generate torch `Dataloader` object from data directory path.
This provides clear tensor input representations for compatible models.
Returns a `Dataloader`
* **dir_path** - String path to directory of images you'd like to process
* **mini_batch_size** - Batch size for inference
* **use_gpu** - Bool for using gpu or cpu. If set True but no gpu devices available, model will default to using cpu
"""
if use_gpu:
if torch.cuda.is_available():
device = torch.device("cuda")
else:
logger.info("GPU not available")
device = torch.device("cpu")
else:
device = torch.device("cpu")
dataset = self.process(dir_path=dir_path, transforms=transforms)
# Instead of a tensor batch, the lambda collate_fn will provide a list batch
dataloader = DataLoader(
dataset,
batch_size=mini_batch_size,
collate_fn=lambda x: [[t.to(device) for t in self._od_collate_fn(x)]],
)
return dataloader
process_output_batch(self, outputs, dataset)
¶
Process output of object detection model into human legible results.
Outputs from FasterRCNNModule
Returns batched results of list of list of tuples containing boxed images in tensor and numpy format
- outputs - List of batch output tensors from a model's forward pass
- dataset - Corresponding dataset with originial images matched with model outputs
Source code in fastnn/processors/cv/object_detection.py
def process_output_batch(
self, outputs: List[List[torch.Tensor]], dataset: Dataset
) -> List[List[Tuple[torch.Tensor, np.array]]]:
"""Process output of object detection model into human legible results.
Outputs from `FasterRCNNModule`
Returns batched results of list of list of tuples containing boxed images in tensor and numpy format
* **outputs** - List of batch output tensors from a model's forward pass
* **dataset** - Corresponding dataset with originial images matched with model outputs
"""
# Labeled Images
results = []
for idx, out in enumerate(outputs):
labeled_images = []
for label_idx in range(1, len(out), 3):
labels = [self.label_strings[o] for o in out[label_idx]]
unique_labels = set(labels)
label_colors_map = {}
for label in unique_labels:
label_colors_map[label] = tuple(
np.random.choice(range(256), size=3)
)
label_colors = [label_colors_map[label] for label in labels]
output_tensor, output_numpy = self.draw_bounding_boxes(
ConvertImageDtype(torch.uint8)(
dataset[idx * (len(out) // 3) + label_idx // 3]
),
out[label_idx - 1],
labels=labels,
colors=label_colors,
)
labeled_images.append((output_tensor, output_numpy))
results.append(labeled_images)
return results