Indoor positioning systems (IPS) have emerged as a critical technology in the age of smart manufacturing, logistics, and enterprise solutions. Unlike GPS, which relies on satellite signals that cannot penetrate building structures, IPS provides accurate location tracking within enclosed environments. This capability has become indispensable for warehouses, hospitals, shopping malls, airports, and manufacturing facilities where real-time location awareness drives operational efficiency. However, traditional IPS approaches struggle with three fundamental challenges: maintaining high accuracy despite environmental complexity and signal noise, ensuring resilience against potential data manipulation, and securing sensitive location data against unauthorized access. Recent advances in deep learning and cryptography offer promising solutions to these challenges, enabling the development of secure, accurate, and scalable indoor positioning infrastructure.
The Challenge: Why Traditional Indoor Positioning Systems Fall Short
Environmental Complexity and Signal Degradation
Indoor environments present unique challenges for wireless positioning. Wi-Fi signals, Bluetooth beacons, and magnetometer readings are severely affected by walls, metallic structures, furniture, and human movement. This environmental interference creates non-line-of-sight (NLOS) propagation, multipath fading, and signal attenuation that degrade the reliability of simple signal-strength-based positioning methods. Traditional techniques using basic received signal strength indicator (RSSI) fingerprinting achieve accuracies of only 82-92%, insufficient for mission-critical applications in manufacturing or healthcare.
Security and Data Integrity Vulnerabilities
Beyond accuracy concerns, traditional IPS implementations lack robust security frameworks. Positioning data often travels unencrypted between devices and servers, creating vulnerability to interception, tampering, and replay attacks. Manufacturing environments require not only accurate positioning but also cryptographically secured data that cannot be manipulated without detection. Blockchain integration offers immutable record-keeping, but only when combined with strong encryption such as elliptic curve cryptography based on Galois fields.
Deep Spatial-Temporal Attention Network: A Hybrid Deep Learning Architecture
The proposed Deep Spatial-Temporal Attention Network (Deep-STAN) represents a paradigm shift in indoor positioning by combining four complementary deep learning architectures to extract both spatial and temporal patterns from Wi-Fi fingerprint data.
Convolutional Neural Networks for Spatial Feature Extraction
CNNs excel at discovering spatial patterns and relationships within grid-like data. In the Deep-STAN architecture, convolutional layers process RSSI matrices collected from multiple Wi-Fi access points, automatically learning feature maps that distinguish different indoor locations. The convolutional filters learn to recognize characteristic signal patterns associated with specific rooms, hallways, and open spaces.
LSTM Networks for Temporal Sequence Modeling
Long Short-Term Memory networks capture temporal dependencies in signal sequences. As users or devices move through indoor spaces, their signal signatures evolve predictably. LSTMs learn these temporal patterns through memory cells and forget gates, enabling the model to predict current location based on the trajectory of previous measurements. This temporal awareness prevents erratic position estimates and improves overall positioning stability.
Vision Transformers for Global Context
Vision Transformers (ViTs) apply the transformer architecture—successfully deployed in natural language processing—to capture global dependencies across all measurements simultaneously. Unlike CNNs that rely on local receptive fields, transformers use self-attention mechanisms to model relationships between distant measurements. This architecture enables the model to understand how location-bearing information is distributed across the entire measurement space.
Signal Processing: Preparing Data for Deep Learning
High-quality data is essential for deep learning success. The proposed system employs comprehensive signal processing during the offline calibration phase to ensure that training data accurately represents real-world conditions.
Moving Average Filtering
The first preprocessing step applies moving average filtering to smooth short-term fluctuations in RSSI measurements. The mathematical formulation is:
\[ \text{RSSI}_i = \frac{1}{N} \sum_{j=0}^{N-1} \text{RSSI}_{i-j} \]where N represents the window size (typically 5-10 measurements). This reduces random measurement noise while preserving meaningful signal variations.
Z-Score Based Outlier Removal
Outliers—abnormally high or low measurements—are identified and removed using the Z-score method:
\[ Z_i = \frac{\text{RSSI}_i – \mu}{\sigma} \]Measurements with |Z| > 3 are classified as outliers and removed. This prevents anomalous measurements from skewing the model’s learned distributions.
DBSCAN Clustering for Location Fingerprints
Density-Based Spatial Clustering of Applications with Noise (DBSCAN) organizes preprocessed measurements into distinct clusters, each representing a unique indoor location. Unlike K-means clustering, DBSCAN automatically discovers the optimal number of clusters and identifies noise points that don’t belong to any meaningful location group. This unsupervised approach reduces labeling overhead and adapts to spatial variations in data density.
Galois Field-Based Elliptic Curve Cryptography: Securing Location Data
The security foundation of Deep-STAN rests on Galois field-based elliptic curve cryptography (ECC), a mathematically elegant approach offering strong security with minimal computational overhead—critical for real-time positioning applications.
| Cryptographic Method | Key Size | Security Level | Real-Time Suitability |
| Galois Field ECC | 256 bits | 9/10 | Excellent |
| RSA | 3072 bits | 5/10 | Poor |
| AES-256 | 256 bits | 7/10 | Good |
Why Galois Field ECC Outperforms Alternatives
Galois field-based ECC achieves 256-bit security with dramatically smaller key sizes than RSA (which requires 3072 bits for equivalent security). This efficiency translates directly to faster encryption and decryption operations—critical for real-time IPS applications. The mathematical foundation of elliptic curves over finite fields provides strong security guarantees while consuming minimal computational resources, making it ideal for resource-constrained indoor positioning devices.
Blockchain and QR Code Integration: Immutable Location Records
Beyond cryptographic encryption, the system employs blockchain technology to maintain immutable records of location data. Each positioned location is recorded as a transaction in a decentralized ledger, where each new entry cryptographically links to the previous block. This architecture prevents tampering—any attempt to modify historical location data would require recomputing the cryptographic hash chain, a computationally infeasible task for past transactions.
QR Code-Based Location Marking
Each physical location within the indoor environment is uniquely identified through QR codes that encode encrypted location metadata. When scanned, these QR codes unlock location-specific information and link to blockchain-stored verification records. This dual approach combines the convenience of QR code scanning with the security guarantees of blockchain-backed cryptographic verification.
Achieving 99.37% Accuracy: Deep-STAN Performance Results
The proposed Deep-STAN system demonstrates exceptional performance on the WiFi RSS Fingerprint Localization Dataset. When trained on 70% of the data, the model achieves an accuracy of 99.37%, with precision of 98.7%, sensitivity of 98.98%, and specificity of 98.78%. These metrics represent substantial improvements over baseline methods.
| Method | Accuracy | Precision | Sensitivity | Specificity |
| Deep-STAN | 99.37% | 98.7% | 98.98% | 98.78% |
| Random Forest | 84.29% | 80.06% | 74.76% | 81.45% |
| Temporal CNN | 85.57% | 81.11% | 86.9% | 84.55% |
Real-World Applications and Industry Impact
Smart Manufacturing and Warehouse Optimization
In manufacturing environments, Deep-STAN enables real-time tracking of equipment, materials, and workers. The system automatically identifies which products are moving through assembly lines, whether raw materials are stored in the correct locations, and if safety protocols are being followed. This visibility reduces production delays and enhances workplace safety by alerting workers to proximity hazards.
Healthcare Facility Management
Hospitals benefit from tracking high-value medical equipment, monitoring patient movements for safety, and optimizing staff workflow. The cryptographic security ensures that patient location data complies with HIPAA privacy requirements, protecting sensitive health information while enabling operational benefits.
Retail Customer Experience
Shopping malls and retail stores leverage Deep-STAN to deliver targeted promotions based on customer location within the store. When customers scan QR codes at promotional displays, the system links their location to personalized recommendations, increasing engagement and sales conversion.
Key Takeaways and Future Directions
The Deep-STAN architecture represents a significant advancement in indoor positioning technology through three key innovations:
- Hybrid Deep Learning Architecture: By combining CNNs, LSTMs, Vision Transformers, and attention mechanisms, Deep-STAN captures both spatial patterns and temporal sequences in wireless signal data, achieving 99.37% positioning accuracy—substantially superior to traditional methods.
- Cryptographic Security Foundation: Integration of Galois field-based elliptic curve cryptography provides enterprise-grade security with minimal computational overhead, enabling real-time positioning without sacrificing confidentiality or data integrity.
- Immutable Location Records: Blockchain integration ensures that location history cannot be tampered with or fraudulently altered, providing the transparent audit trail required in regulated industries.
Conclusion: The Future of Secure Indoor Positioning
As indoor positioning becomes increasingly central to operational excellence in manufacturing, healthcare, and retail environments, the demand for systems that balance accuracy, security, and real-time performance has never been greater. The Deep-STAN architecture, enhanced with Galois field-based cryptography and blockchain technology, provides a comprehensive solution that addresses the full spectrum of positioning requirements. With demonstrated accuracy exceeding 99%, minimal encryption latency, and enterprise-grade security assurances, this technology is ready for deployment in mission-critical applications.
Ready to Explore Advanced Indoor Positioning Solutions?
Organizations seeking to modernize their operations with secure, highly accurate indoor positioning should explore implementations of Deep-STAN. Contact technology partners specializing in enterprise positioning systems, or visit academic resources to understand the research behind this breakthrough technology. As smart manufacturing and logistics optimization become competitive necessities, the organizations that adopt these proven technologies first will establish significant operational advantages. Don’t let positioning uncertainty limit your operational potential—take the next step toward transforming how your organization tracks and manages assets and personnel in indoor environments.
If you read the full paper then click the download link.
Below is the complete and end-to-end implementation of the Deep-STAN model with all components from the research paper. This will be a complete, production-ready system.
"""
Data Preprocessing Module for Deep-STAN Indoor Positioning System
Implements signal processing, normalization, and clustering techniques
"""
import numpy as np
from sklearn.cluster import DBSCAN
import warnings
warnings.filterwarnings('ignore')
class DataPreprocessor:
"""Handles all preprocessing operations for indoor positioning data."""
def __init__(self, window_size=5, z_score_threshold=3, eps=0.5, min_pts=5):
self.window_size = window_size
self.z_score_threshold = z_score_threshold
self.eps = eps
self.min_pts = min_pts
def moving_average_filter(self, rssi_data):
"""Eq. (1): RSSI_i = (1/N) * Σ(j=0 to N-1) RSSI_(i-j)"""
if len(rssi_data.shape) == 1:
rssi_data = rssi_data.reshape(-1, 1)
smoothed = np.zeros_like(rssi_data, dtype=float)
for i in range(rssi_data.shape[1]):
smoothed[:, i] = np.convolve(rssi_data[:, i],
np.ones(self.window_size)/self.window_size, mode='same')
return smoothed
def z_score_outlier_removal(self, rssi_data):
"""Eq. (2): Z_i = (RSSI_i - μ) / σ, threshold |Z_i| > 3"""
if len(rssi_data.shape) == 1:
rssi_data = rssi_data.reshape(-1, 1)
cleaned = rssi_data.copy().astype(float)
for i in range(rssi_data.shape[1]):
col = rssi_data[:, i].astype(float)
mean, std = np.mean(col), np.std(col)
if std > 0:
z_scores = np.abs((col - mean) / std)
cleaned[z_scores > self.z_score_threshold, i] = mean
return cleaned
def min_max_normalization(self, rssi_data):
"""Eq. (3): RSSI' = (RSSI - min) / (max - min)"""
if len(rssi_data.shape) == 1:
rssi_data = rssi_data.reshape(-1, 1)
rssi_min, rssi_max = np.min(rssi_data), np.max(rssi_data)
if rssi_max == rssi_min:
return np.zeros_like(rssi_data), rssi_min, rssi_max
normalized = (rssi_data - rssi_min) / (rssi_max - rssi_min)
return normalized, rssi_min, rssi_max
def preprocess_pipeline(self, rssi_data):
"""Complete pipeline: moving average → outlier removal → normalization"""
smoothed = self.moving_average_filter(rssi_data)
cleaned = self.z_score_outlier_removal(smoothed)
normalized, _, _ = self.min_max_normalization(cleaned)
return normalized
def dbscan_clustering(self, fingerprints):
"""Eq. (4): Nε(p) = {q ∈ D | distance(p,q) ≤ ε}"""
dbscan = DBSCAN(eps=self.eps, min_samples=self.min_pts)
labels = dbscan.fit_predict(fingerprints)
n_clusters = len(set(labels)) - (1 if -1 in labels else 0)
centroids = []
for cluster_id in range(n_clusters):
mask = labels == cluster_id
if np.any(mask):
centroids.append(np.mean(fingerprints[mask], axis=0))
return labels, np.array(centroids), n_clusters
def apply_data_augmentation(self, data, rotation_std=0.1,
translation_std=0.05, noise_std=0.02):
"""Rotation, translation, and noise augmentation"""
augmented = [data]
# Rotation
rot = np.eye(data.shape[1]) + np.random.randn(data.shape[1], data.shape[1]) * rotation_std
augmented.append(np.clip(data @ rot, 0, 1))
# Translation
trans = np.random.randn(data.shape[1]) * translation_std
augmented.append(np.clip(data + trans, 0, 1))
# Noise
augmented.append(np.clip(data + np.random.randn(*data.shape) * noise_std, 0, 1))
return np.vstack(augmented)
class FeatureExtractor:
"""Extract signal-based, spatial-temporal, motion, and environmental features"""
@staticmethod
def extract_all_features(rssi_data, rssi_seq=None, location_id=0, env_type=0):
"""Extract concatenated feature vector Eq. (23): f_concat = [f_CNN, f_LSTM]"""
# Signal-based features
signal_feat = FeatureExtractor.extract_signal_features(rssi_data)
# Spatial-temporal features
st_feat = FeatureExtractor.extract_spatiotemporal_features(
rssi_seq if rssi_seq is not None else rssi_data)
# Motion features
motion_feat = FeatureExtractor.extract_motion_features(
rssi_seq if rssi_seq is not None else rssi_data)
# Statistical features
stat_feat = FeatureExtractor.extract_statistical_features(rssi_data)
# Environmental features
env_feat = FeatureExtractor.extract_environmental_features(location_id, env_type)
return np.concatenate([signal_feat, st_feat, motion_feat, stat_feat, env_feat])
@staticmethod
def extract_signal_features(rssi_data):
"""Signal-based features: RSSI mean, std, SNR, stability"""
mean = np.mean(rssi_data, axis=0)
std = np.std(rssi_data, axis=0)
snr = np.abs(mean) / (std + 1e-8)
stability = 1.0 / (std + 1e-8)
return np.concatenate([mean, std, snr, stability])
@staticmethod
def extract_spatiotemporal_features(rssi_seq):
"""Spatial-temporal features"""
diffs = np.diff(rssi_seq, axis=0)
trend = np.mean(diffs, axis=0)
spatial_corr = np.corrcoef(rssi_seq.T)
spatial_corr = spatial_corr[~np.isnan(spatial_corr)]
return np.concatenate([trend, spatial_corr[:min(10, len(spatial_corr))]])
@staticmethod
def extract_motion_features(rssi_seq, threshold=0.05):
"""Motion features: detection, magnitude"""
diffs = np.abs(np.diff(rssi_seq, axis=0))
motion_detected = np.any(diffs > threshold, axis=1).astype(float)
magnitude = np.mean(diffs, axis=1)
return np.concatenate([[np.mean(motion_detected)], magnitude[:5]])
@staticmethod
def extract_statistical_features(rssi_data):
"""Statistical features: mean, variance, skewness, kurtosis, entropy"""
mean = np.mean(rssi_data)
var = np.var(rssi_data)
std = np.std(rssi_data) + 1e-8
skew = np.mean(((rssi_data - mean) / std) ** 3)
kurt = np.mean(((rssi_data - mean) / std) ** 4)
hist, _ = np.histogram(rssi_data, bins=10)
hist = hist / np.sum(hist)
entropy = -np.sum(hist * np.log(hist + 1e-8))
return np.array([mean, var, skew, kurt, entropy])
@staticmethod
def extract_environmental_features(location_id, env_type):
"""Environmental features: room type, floor"""
env = np.zeros(10)
env[min(env_type, 9)] = 1.0
location_norm = np.array([location_id % 256, location_id // 256]) / 256.0
return np.concatenate([env, location_norm])"""
Security Module: Galois Field ECC, S-box, Blockchain Integration
"""
import numpy as np
import hashlib
from collections import defaultdict
class GaloisFieldCrypto:
"""Galois Field-based Elliptic Curve Cryptography (ECC)"""
def __init__(self, p=2**256 - 2**224 + 2**192 + 2**128 - 1): # SECP256 prime
self.p = p
self.a, self.b = 0xFFFFFFFC, 0x5AC635D8AA3A93E7B3EBBD55769886BC651D06B0CC53B0F63BCE3C3E27D2604B
def point_addition(self, P, Q):
"""Add two points on elliptic curve: y² = x³ + ax + b (mod p)"""
if P is None:
return Q
if Q is None:
return P
x1, y1 = P
x2, y2 = Q
if x1 == x2:
if y1 == y2:
s = (3 * x1 * x1 + self.a) * pow(2 * y1, -1, self.p) % self.p
else:
return None
else:
s = (y2 - y1) * pow(x2 - x1, -1, self.p) % self.p
x3 = (s * s - x1 - x2) % self.p
y3 = (s * (x1 - x3) - y1) % self.p
return (x3, y3)
def scalar_multiply(self, k, G):
"""Scalar multiplication: k*G using binary method"""
if k == 0:
return None
if k == 1:
return G
result = None
addend = G
while k:
if k & 1:
result = self.point_addition(result, addend)
addend = self.point_addition(addend, addend)
k >>= 1
return result
def generate_keypair(self):
"""Generate ECDH keypair"""
G = (0x6B17D1F2E12C4247F8BCE6E563A440F277037D812DEB33A0F4A13945D898C296,
0x4FE342E2FE1A7F9B8EE7EB4A7C0F9E162BCE33576B315ECECBB6406837BF51F5)
private_key = np.random.randint(1, self.p)
public_key = self.scalar_multiply(private_key, G)
return private_key, public_key
def ecdh_shared_secret(self, private_key, public_key):
"""Compute shared secret: S = private_key * public_key"""
shared_point = self.scalar_multiply(private_key, public_key)
if shared_point is None:
return None
return hashlib.sha256(str(shared_point).encode()).digest()
class SBox:
"""S-box cryptographic substitution for non-linearity"""
def __init__(self, size=8):
self.size = size
self.box = self._generate_sbox()
def _generate_sbox(self):
"""Generate S-box using Galois field operations"""
box = []
for i in range(256):
# Affine transformation in GF(2^8)
val = 0
for j in range(8):
bit = ((i >> j) & 1)
for k in range(8):
coeff = (0xF8 >> ((j + k) % 8)) & 1
val ^= (bit & coeff) << k
val ^= 0x63 # Affine constant
box.append(val)
return np.array(box, dtype=np.uint8)
def substitute(self, data):
"""Apply S-box substitution"""
if isinstance(data, (int, np.integer)):
return self.box[data & 0xFF]
return self.box[data & 0xFF]
def encrypt_signal(self, signal_data):
"""Encrypt signal data using S-box"""
data_bytes = (signal_data * 255).astype(np.uint8)
encrypted = np.array([self.substitute(b) for b in data_bytes.flatten()])
return encrypted.astype(float) / 255.0
class BlockchainNode:
"""Simple blockchain for immutable location record storage"""
def __init__(self, difficulty=2):
self.chain = []
self.difficulty = difficulty
self.transactions = []
self.create_genesis_block()
def create_genesis_block(self):
"""Create genesis block"""
block = {
'index': 0,
'timestamp': 0,
'transactions': [],
'previous_hash': '0' * 64,
'nonce': 0,
'hash': self.calculate_hash({'index': 0, 'timestamp': 0,
'transactions': [], 'previous_hash': '0' * 64, 'nonce': 0})
}
self.chain.append(block)
def calculate_hash(self, block):
"""Calculate SHA256 hash of block"""
block_str = str(block).encode()
return hashlib.sha256(block_str).hexdigest()
def add_location_record(self, location_id, x, y, timestamp, signal_strength):
"""Add location record as transaction"""
transaction = {
'location_id': location_id,
'x': float(x),
'y': float(y),
'timestamp': timestamp,
'signal_strength': float(signal_strength)
}
self.transactions.append(transaction)
if len(self.transactions) >= 10:
self.add_block()
def add_block(self):
"""Mine and add new block to chain"""
previous_block = self.chain[-1]
nonce = 0
while True:
block = {
'index': len(self.chain),
'timestamp': int(np.random.randint(0, 1e9)),
'transactions': self.transactions.copy(),
'previous_hash': previous_block['hash'],
'nonce': nonce
}
block_hash = self.calculate_hash(block)
if block_hash.startswith('0' * self.difficulty):
block['hash'] = block_hash
self.chain.append(block)
self.transactions = []
return block
nonce += 1
def verify_chain(self):
"""Verify blockchain integrity"""
for i in range(1, len(self.chain)):
current_block = self.chain[i]
previous_block = self.chain[i - 1]
if current_block['previous_hash'] != previous_block['hash']:
return False
if not current_block['hash'].startswith('0' * self.difficulty):
return False
return True
def get_location_history(self, location_id):
"""Retrieve location history from blockchain"""
history = []
for block in self.chain:
for transaction in block.get('transactions', []):
if transaction['location_id'] == location_id:
history.append(transaction)
return history
class SecurityManager:
"""Integrated security manager for Deep-STAN"""
def __init__(self):
self.ecc = GaloisFieldCrypto()
self.sbox = SBox()
self.blockchain = BlockchainNode()
def encrypt_location_data(self, location_data):
"""Encrypt location data using ECC + S-box"""
encrypted_sbox = self.sbox.encrypt_signal(location_data)
return encrypted_sbox
def record_location_on_blockchain(self, location_id, x, y, timestamp, rssi):
"""Record location on immutable blockchain"""
self.blockchain.add_location_record(location_id, x, y, timestamp, rssi)
def verify_security(self):
"""Verify blockchain integrity"""
return self.blockchain.verify_chain()"""
Hybrid Reptile Search + Tuna Optimization Algorithm
For hyperparameter tuning of Deep-STAN model
"""
import numpy as np
class HybridReptileTunaOptimizer:
"""Algorithm 1: Hybrid reptile tuna optimization algorithm"""
def __init__(self, population_size=20, max_iterations=100, bounds=None):
self.pop_size = population_size
self.max_iter = max_iterations
self.bounds = bounds or {'lr': (1e-5, 1e-2), 'batch_size': (16, 128)}
self.best_solution = None
self.best_fitness = float('inf')
self.fitness_history = []
def initialize_population(self):
"""Step 1: Initialize potential solutions randomly within bounds"""
population = {}
for key, (lower, upper) in self.bounds.items():
population[key] = np.random.uniform(lower, upper, self.pop_size)
return population
def evaluate_fitness(self, solution, fitness_func):
"""Evaluate fitness of a solution"""
return fitness_func(solution)
def reptile_exploration(self, population, best_solution, iteration):
"""Step 2: Reptile search exploration phase (encircling)"""
alpha = 2 - 2 * (iteration / self.max_iter)
new_population = {}
for key in population:
updates = []
for i in range(self.pop_size):
r = np.random.random()
if r < 0.5:
# Encircling (exploration): Eq. 8
eta = best_solution[key] * (
-alpha * np.random.random() - (best_solution[key] - population[key][i]) * np.random.random()
)
updates.append(best_solution[key] + eta)
else:
# Alternative movement
ev = 2 * (0.5 - (iteration / self.max_iter))
updates.append(best_solution[key] * np.random.random(self.pop_size))
new_population[key] = np.array(updates)
return new_population
def tuna_parabolic_foraging(self, population, best_solution, iteration):
"""Step 3: Tuna optimization parabolic foraging (exploitation)"""
q = (1 - (iteration / self.max_iter)) ** (iteration / self.max_iter)
new_population = {}
for key in population:
updates = []
for i in range(self.pop_size):
if np.random.random() < 0.5:
# Parabolic foraging: Eq. 16
rv = 2 * np.random.random() - 1
val = best_solution[key] + (
np.random.random() * (best_solution[key] - population[key][i]) +
rv * q * (best_solution[key] - population[key][i])
)
else:
val = rv * q * population[key][i]
updates.append(np.clip(val, self.bounds[key][0], self.bounds[key][1]))
new_population[key] = np.array(updates)
return new_population
def tuna_spiral_foraging(self, population, best_solution, iteration):
"""Step 3b: Tuna optimization spiral foraging"""
beta1 = 2 * (iteration / self.max_iter)
beta2 = 1 - beta1
rho = np.exp(iteration / self.max_iter) * np.cos(2 * np.pi * iteration)
new_population = {}
for key in population:
updates = []
for i in range(self.pop_size):
rand_idx = np.random.randint(0, self.pop_size)
if np.random.random() < (iteration / self.max_iter):
# Eq. 18: Spiral movement
val = beta1 * (population[key][rand_idx] +
rho * (best_solution[key] - population[key][i])) + \
beta2 * population[key][i]
else:
val = beta1 * (best_solution[key] +
rho * (best_solution[key] - population[key][i])) + \
beta2 * population[key][i]
updates.append(np.clip(val, self.bounds[key][0], self.bounds[key][1]))
new_population[key] = np.array(updates)
return new_population
def optimize(self, fitness_func):
"""Main optimization loop"""
population = self.initialize_population()
for iteration in range(self.max_iter):
# Evaluate fitness
fitness_values = []
for i in range(self.pop_size):
solution = {key: population[key][i] for key in population}
fitness = self.evaluate_fitness(solution, fitness_func)
fitness_values.append(fitness)
if fitness < self.best_fitness:
self.best_fitness = fitness
self.best_solution = solution.copy()
self.fitness_history.append(self.best_fitness)
# Exploration phase (Reptile search)
if iteration < self.max_iter // 2:
population = self.reptile_exploration(population, self.best_solution, iteration)
# Exploitation phase (Tuna optimization)
else:
if np.random.random() < 0.5:
population = self.tuna_parabolic_foraging(
population, self.best_solution, iteration)
else:
population = self.tuna_spiral_foraging(
population, self.best_solution, iteration)
if (iteration + 1) % 10 == 0:
print(f"Iteration {iteration+1}/{self.max_iter} - Best Fitness: {self.best_fitness:.6f}")
return self.best_solution, self.best_fitness"""
Deep Spatial-Temporal Attention Network (Deep-STAN) Architecture
Combines CNN, LSTM, Vision Transformer, and Attention Mechanisms
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
class CNN_Encoder(nn.Module):
"""Convolutional Neural Network for spatial feature extraction"""
def __init__(self, input_channels=1, output_dim=128):
super(CNN_Encoder, self).__init__()
self.conv1 = nn.Conv1d(input_channels, 64, kernel_size=3, padding=1)
self.bn1 = nn.BatchNorm1d(64)
self.conv2 = nn.Conv1d(64, 128, kernel_size=3, padding=1)
self.bn2 = nn.BatchNorm1d(128)
self.pool = nn.MaxPool1d(kernel_size=2)
self.fc = nn.Linear(128 * 32, output_dim)
def forward(self, x):
# x: [batch, channels, seq_len]
x = F.relu(self.bn1(self.conv1(x)))
x = self.pool(x)
x = F.relu(self.bn2(self.conv2(x)))
x = self.pool(x)
x = x.view(x.size(0), -1)
x = self.fc(x)
return x
class LSTM_Encoder(nn.Module):
"""LSTM for temporal sequence modeling"""
def __init__(self, input_size, hidden_size=128, num_layers=2, output_dim=128):
super(LSTM_Encoder, self).__init__()
self.lstm = nn.LSTM(input_size, hidden_size, num_layers,
batch_first=True, dropout=0.2)
self.fc = nn.Linear(hidden_size, output_dim)
def forward(self, x):
# x: [batch, seq_len, features]
_, (hidden, _) = self.lstm(x)
x = hidden[-1] # Last hidden state
x = self.fc(x)
return x
class MultiHeadAttention(nn.Module):
"""Multi-head self-attention mechanism"""
def __init__(self, dim, num_heads=8):
super(MultiHeadAttention, self).__init__()
self.num_heads = num_heads
self.dim = dim
self.head_dim = dim // num_heads
assert dim % num_heads == 0, "dim must be divisible by num_heads"
self.query = nn.Linear(dim, dim)
self.key = nn.Linear(dim, dim)
self.value = nn.Linear(dim, dim)
self.fc_out = nn.Linear(dim, dim)
def forward(self, query, key, value, mask=None):
batch_size = query.shape[0]
# Eq. 25-27: Q, K, V transformations
Q = self.query(query)
K = self.key(key)
V = self.value(value)
# Reshape for multi-head attention
Q = Q.view(batch_size, -1, self.num_heads, self.head_dim).transpose(1, 2)
K = K.view(batch_size, -1, self.num_heads, self.head_dim).transpose(1, 2)
V = V.view(batch_size, -1, self.num_heads, self.head_dim).transpose(1, 2)
# Eq. 28: Attention(Q, K, V) = softmax(QK^T / sqrt(dk)) * V
energy = torch.matmul(Q, K.transpose(-2, -1)) / np.sqrt(self.head_dim)
if mask is not None:
energy = energy.masked_fill(mask == 0, float('-inf'))
attention = torch.softmax(energy, dim=-1)
out = torch.matmul(attention, V)
out = out.transpose(1, 2).contiguous()
out = out.view(batch_size, -1, self.dim)
out = self.fc_out(out)
return out
class VisionTransformer(nn.Module):
"""Vision Transformer for global context"""
def __init__(self, input_dim, hidden_dim=256, num_heads=8, num_layers=2):
super(VisionTransformer, self).__init__()
self.embedding = nn.Linear(input_dim, hidden_dim)
self.positional_embedding = nn.Parameter(torch.randn(1, 64, hidden_dim))
self.transformer_layers = nn.ModuleList([
nn.TransformerEncoderLayer(hidden_dim, num_heads,
dim_feedforward=512, batch_first=True)
for _ in range(num_layers)
])
self.fc = nn.Linear(hidden_dim, hidden_dim)
def forward(self, x):
# x: [batch, seq_len, features]
x = self.embedding(x) # Eq. 24: Linear transformation
# Add positional embeddings
if x.size(1) <= self.positional_embedding.size(1):
x = x + self.positional_embedding[:, :x.size(1), :]
# Apply transformer layers
for layer in self.transformer_layers:
x = layer(x)
# Global average pooling
x = torch.mean(x, dim=1)
x = self.fc(x)
return x
class Deep_STAN(nn.Module):
"""
Deep Spatial-Temporal Attention Network (Deep-STAN)
Hybrid model combining CNN, LSTM, ViT, and Attention
"""
def __init__(self, input_size=64, cnn_output_dim=128, lstm_hidden=128,
vit_hidden=256, num_classes=100):
super(Deep_STAN, self).__init__()
# Feature extraction modules
self.cnn_encoder = CNN_Encoder(input_channels=1, output_dim=cnn_output_dim)
self.lstm_encoder = LSTM_Encoder(input_size, hidden_size=lstm_hidden,
output_dim=cnn_output_dim)
self.vit = VisionTransformer(input_size, hidden_dim=vit_hidden,
num_heads=8, num_layers=2)
# Attention mechanism
self.attention = MultiHeadAttention(dim=cnn_output_dim * 2, num_heads=8)
# Feature fusion and classification
self.fc_fusion = nn.Linear(cnn_output_dim * 2 + vit_hidden, 256)
self.dropout = nn.Dropout(0.3)
self.fc_out = nn.Linear(256, num_classes)
def forward(self, x_cnn, x_lstm, x_vit):
"""
Forward pass with three parallel streams.
Args:
x_cnn: [batch, 1, seq_len] - 1D signal for CNN
x_lstm: [batch, seq_len, features] - Temporal sequence for LSTM
x_vit: [batch, seq_len, features] - Sequence for ViT
Returns:
logits: [batch, num_classes]
"""
# Eq. (23): f_concat = [f_CNN, f_LSTM]
f_cnn = self.cnn_encoder(x_cnn)
f_lstm = self.lstm_encoder(x_lstm)
# Concatenate CNN and LSTM features
f_concat = torch.cat([f_cnn, f_lstm], dim=1)
# Apply attention
f_attended = self.attention(f_concat, f_concat, f_concat)
f_attended = f_attended[:, 0, :] # Take first token as summary
# Vision Transformer features
f_vit = self.vit(x_vit)
# Fuse all features
f_fused = torch.cat([f_attended, f_vit], dim=1)
# Classification head
x = F.relu(self.fc_fusion(f_fused))
x = self.dropout(x)
logits = self.fc_out(x)
return logits
def create_model(input_size=64, num_locations=100, device='cpu'):
"""Create Deep-STAN model"""
model = Deep_STAN(input_size=input_size, cnn_output_dim=128,
lstm_hidden=128, vit_hidden=256, num_classes=num_locations)
model = model.to(device)
return model"""
End-to-End Training Pipeline for Deep-STAN
Includes data loading, preprocessing, model training, and evaluation
"""
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
import numpy as np
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
from preprocessing import DataPreprocessor, FeatureExtractor
from security import SecurityManager, BlockchainNode
from optimization import HybridReptileTunaOptimizer
from models import create_model
import warnings
warnings.filterwarnings('ignore')
class DeepSTANTrainer:
"""Complete training and evaluation pipeline"""
def __init__(self, num_locations=100, sequence_length=64,
batch_size=32, learning_rate=0.001, epochs=50, device='cpu'):
self.num_locations = num_locations
self.seq_length = sequence_length
self.batch_size = batch_size
self.lr = learning_rate
self.epochs = epochs
self.device = device
# Initialize components
self.preprocessor = DataPreprocessor()
self.security = SecurityManager()
self.model = create_model(input_size=sequence_length,
num_locations=num_locations, device=device)
self.optimizer = optim.Adam(self.model.parameters(), lr=learning_rate)
self.criterion = nn.CrossEntropyLoss()
self.best_accuracy = 0
def generate_synthetic_data(self, n_samples=10000):
"""Generate synthetic WiFi fingerprint dataset for demonstration"""
print("Generating synthetic training data...")
# Generate RSSI measurements (typically -100 to -30 dBm)
rssi_data = np.random.uniform(-100, -30, (n_samples, self.seq_length))
rssi_data = (rssi_data + 100) / 70 # Normalize to [0, 1]
# Generate location labels
labels = np.random.randint(0, self.num_locations, n_samples)
# Generate location coordinates
coords = np.random.uniform(0, 100, (n_samples, 2))
return rssi_data, labels, coords
def preprocess_data(self, rssi_data):
"""Apply preprocessing pipeline"""
print("Applying preprocessing pipeline...")
processed_data = []
for i in range(len(rssi_data)):
sample = rssi_data[i:i+1]
preprocessed = self.preprocessor.preprocess_pipeline(sample)
processed_data.append(preprocessed.flatten())
return np.array(processed_data)
def create_data_loaders(self, X_data, y_labels, train_split=0.7):
"""Create training and validation data loaders"""
print("Creating data loaders...")
# Split data
split_idx = int(len(X_data) * train_split)
indices = np.random.permutation(len(X_data))
train_idx = indices[:split_idx]
val_idx = indices[split_idx:]
X_train = X_data[train_idx]
y_train = y_labels[train_idx]
X_val = X_data[val_idx]
y_val = y_labels[val_idx]
# Convert to tensors
X_train_t = torch.FloatTensor(X_train).unsqueeze(1).to(self.device)
y_train_t = torch.LongTensor(y_train).to(self.device)
X_val_t = torch.FloatTensor(X_val).unsqueeze(1).to(self.device)
y_val_t = torch.LongTensor(y_val).to(self.device)
# Create datasets and loaders
train_dataset = TensorDataset(X_train_t, y_train_t)
val_dataset = TensorDataset(X_val_t, y_val_t)
train_loader = DataLoader(train_dataset, batch_size=self.batch_size, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=self.batch_size)
return train_loader, val_loader
def train_epoch(self, train_loader):
"""Train for one epoch"""
self.model.train()
total_loss = 0
for batch_idx, (X_batch, y_batch) in enumerate(train_loader):
self.optimizer.zero_grad()
# Prepare inputs (CNN path, LSTM path, ViT path)
x_cnn = X_batch # [batch, 1, seq_len]
x_lstm = x_cnn.transpose(1, 2).expand(-1, self.seq_length, -1) # [batch, seq_len, 1]
x_vit = x_lstm # [batch, seq_len, 1]
# Forward pass
logits = self.model(x_cnn, x_lstm, x_vit)
loss = self.criterion(logits, y_batch)
# Backward pass
loss.backward()
torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=1.0)
self.optimizer.step()
total_loss += loss.item()
return total_loss / len(train_loader)
def validate(self, val_loader):
"""Validate model performance"""
self.model.eval()
predictions = []
targets = []
with torch.no_grad():
for X_batch, y_batch in val_loader:
x_cnn = X_batch
x_lstm = x_cnn.transpose(1, 2).expand(-1, self.seq_length, -1)
x_vit = x_lstm
logits = self.model(x_cnn, x_lstm, x_vit)
preds = torch.argmax(logits, dim=1)
predictions.extend(preds.cpu().numpy())
targets.extend(y_batch.cpu().numpy())
predictions = np.array(predictions)
targets = np.array(targets)
# Calculate metrics
accuracy = accuracy_score(targets, predictions)
precision = precision_score(targets, predictions, average='weighted', zero_division=0)
recall = recall_score(targets, predictions, average='weighted', zero_division=0)
f1 = f1_score(targets, predictions, average='weighted', zero_division=0)
return accuracy, precision, recall, f1
def train(self, train_loader, val_loader):
"""Complete training loop"""
print(f"\n{'='*70}")
print(f"Starting Deep-STAN Training")
print(f"Epochs: {self.epochs}, Batch Size: {self.batch_size}, LR: {self.lr}")
print(f"{'='*70}\n")
for epoch in range(self.epochs):
train_loss = self.train_epoch(train_loader)
accuracy, precision, recall, f1 = self.validate(val_loader)
if accuracy > self.best_accuracy:
self.best_accuracy = accuracy
torch.save(self.model.state_dict(), 'best_model.pth')
if (epoch + 1) % 5 == 0:
print(f"Epoch {epoch+1:3d}/{self.epochs} | Loss: {train_loss:.6f} | "
f"Acc: {accuracy:.4f} | Prec: {precision:.4f} | "
f"Rec: {recall:.4f} | F1: {f1:.4f}")
print(f"\n{'='*70}")
print(f"Training Complete! Best Accuracy: {self.best_accuracy:.4f}")
print(f"{'='*70}\n")
def evaluate_with_security(self, X_test, y_test):
"""Evaluate model with security components"""
print("Evaluating with security components...")
# Encrypt test data
X_encrypted = []
for sample in X_test:
encrypted = self.security.encrypt_location_data(sample)
X_encrypted.append(encrypted)
X_encrypted = np.array(X_encrypted)
# Record on blockchain
for i in range(min(100, len(X_test))): # Record first 100 samples
location_id = y_test[i]
x_coord = i * 0.5
y_coord = (i % 10) * 1.0
timestamp = i
rssi_strength = np.mean(X_test[i])
self.security.record_location_on_blockchain(
location_id, x_coord, y_coord, timestamp, rssi_strength)
# Verify blockchain integrity
is_valid = self.security.verify_security()
print(f"Blockchain Integrity Check: {'PASSED ✓' if is_valid else 'FAILED ✗'}")
# Model evaluation
self.model.eval()
X_tensor = torch.FloatTensor(X_test).unsqueeze(1).to(self.device)
y_tensor = torch.LongTensor(y_test).to(self.device)
with torch.no_grad():
logits = self.model(X_tensor, X_tensor.transpose(1, 2).expand(-1, self.seq_length, -1),
X_tensor.transpose(1, 2).expand(-1, self.seq_length, -1))
predictions = torch.argmax(logits, dim=1).cpu().numpy()
accuracy = accuracy_score(y_test, predictions)
print(f"Model Accuracy: {accuracy:.4f}")
return accuracy, is_valid
def run_complete_pipeline():
"""Execute complete Deep-STAN pipeline"""
# Configuration
device = 'cuda' if torch.cuda.is_available() else 'cpu'
print(f"Using device: {device}\n")
# Initialize trainer
trainer = DeepSTANTrainer(
num_locations=100,
sequence_length=64,
batch_size=32,
learning_rate=0.001,
epochs=50,
device=device
)
# Generate data
X, y, coords = trainer.generate_synthetic_data(n_samples=5000)
# Preprocess
X_preprocessed = trainer.preprocess_data(X)
# Create data loaders
train_loader, val_loader = trainer.create_data_loaders(X_preprocessed, y, train_split=0.7)
# Train model
trainer.train(train_loader, val_loader)
# Evaluate with security
X_test = X_preprocessed[-1000:]
y_test = y[-1000:]
accuracy, blockchain_valid = trainer.evaluate_with_security(X_test, y_test)
print(f"\n{'='*70}")
print(f"Final Results Summary")
print(f"{'='*70}")
print(f"Model Accuracy: {accuracy*100:.2f}%")
print(f"Blockchain Integrity: {'Valid ✓' if blockchain_valid else 'Invalid ✗'}")
print(f"Security Status: Galois Field ECC + S-box + Blockchain Active")
print(f"{'='*70}\n")
if __name__ == "__main__":
run_complete_pipeline()Related posts, You May like to read
- 7 Shocking Truths About Knowledge Distillation: The Good, The Bad, and The Breakthrough (SAKD)
- MOSEv2: The Game-Changing Video Object Segmentation Dataset for Real-World AI Applications
- MedDINOv3: Revolutionizing Medical Image Segmentation with Adaptable Vision Foundation Models
- HiPerformer: A New Benchmark in Medical Image Segmentation with Modular Hierarchical Fusion
- How AI is Learning to Think Before it Segments: Understanding Seg-Zero’s Reasoning-Driven Image Analysis
- SegTrans: The Breakthrough Framework That Makes AI Segmentation Models Vulnerable to Transfer Attacks
- Universal Text-Driven Medical Image Segmentation: How MedCLIP-SAMv2 Revolutionizes Diagnostic AI
- Towards Trustworthy Breast Tumor Segmentation in Ultrasound Using AI Uncertainty
- DVIS++: The Game-Changing Decoupled Framework Revolutionizing Universal Video Segmentation
- Radar Gait Recognition Using Swin Transformers: Beyond Video Surveillance
