Coverage for src/backend/InvenTree/InvenTree/conversion.py: 38%
139 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""Helper functions for converting between units."""
3import logging
4import re
5from hashlib import md5
6from typing import Optional
8from django.core.exceptions import ValidationError
9from django.utils.translation import gettext_lazy as _
11import pint
12import structlog
14from common.settings import get_global_setting, set_global_setting
15from InvenTree.cache import get_session_cache, set_session_cache
17_UNIT_REG_CACHE_KEY = 'unit_registry_hash'
18_unit_registry = None
19_unit_registry_hash: str = ''
21logger = structlog.get_logger('inventree')
23# Disable log output for Pint library
24logging.getLogger('pint').setLevel(logging.ERROR)
27def can_cache_registry() -> bool:
28 """Return True if it is appropriate to cache the unit registry.
30 Prevent caching under certain conditions (such as database migration)
31 to prevent database access.
32 """
33 import InvenTree.ready
35 return not any([
36 InvenTree.ready.isImportingData(),
37 InvenTree.ready.isRunningBackup(),
38 InvenTree.ready.isRunningMigrations(),
39 InvenTree.ready.isInTestMode(),
40 ])
43def get_unit_registry_hash():
44 """Return a hash representing the current state of the unit registry.
46 We use this to determine if we need to reload the unit registry,
47 due to changes in the database.
48 """
49 # Look in the session cache first (faster, and potentially newer)
50 registry_hash = get_session_cache(_UNIT_REG_CACHE_KEY)
52 if registry_hash is None: 52 ↛ 60line 52 didn't jump to line 60 because the condition on line 52 was always true
53 registry_hash = get_global_setting(
54 '_UNIT_REGISTRY_HASH', create=False, backup_value=''
55 )
57 if registry_hash: 57 ↛ 60line 57 didn't jump to line 60 because the condition on line 57 was always true
58 set_session_cache(_UNIT_REG_CACHE_KEY, registry_hash)
60 return registry_hash
63def set_unit_registry_hash(registry_hash: str):
64 """Save the hash representing the current state of the unit registry.
66 Because most of the registry is static, we only need to consider the
67 CustomUnit entries in the database.
68 """
69 global _unit_registry_hash
70 _unit_registry_hash = registry_hash
72 if not can_cache_registry(): 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true
73 return
75 # Save to both the global settings and the session cache
76 set_global_setting('_UNIT_REGISTRY_HASH', registry_hash)
77 set_session_cache(_UNIT_REG_CACHE_KEY, registry_hash)
80def get_unit_registry():
81 """Return a custom instance of the Pint UnitRegistry."""
82 global _unit_registry
83 global _unit_registry_hash
85 # Cache the unit registry for speedier access
86 if _unit_registry is None:
87 return reload_unit_registry()
89 # Check if the unit registry has changed
90 if can_cache_registry() and _unit_registry_hash != get_unit_registry_hash(): 90 ↛ 91line 90 didn't jump to line 91 because the condition on line 90 was never true
91 logger.info('Unit registry hash has changed, reloading unit registry')
92 return reload_unit_registry()
94 return _unit_registry
97def reload_unit_registry():
98 """Reload the unit registry from the database.
100 This function is called at startup, and whenever the database is updated.
101 """
102 import time
104 t_start = time.time()
106 global _unit_registry
108 _unit_registry = None
110 reg = pint.UnitRegistry(autoconvert_offset_to_baseunit=True)
112 # Aliases for temperature units
113 reg.define('@alias degC = Celsius')
114 reg.define('@alias degF = Fahrenheit')
115 reg.define('@alias degK = Kelvin')
117 # Override R as ohm (pint defines R as an SI prefix by default)
118 reg.define('R = ohm')
120 # Define some "standard" additional units
121 reg.define('piece = 1')
122 reg.define('each = 1 = ea')
123 reg.define('dozen = 12 = dz')
124 reg.define('hundred = 100')
125 reg.define('thousand = 1000')
127 # Allow for custom units to be defined in the database
128 # Calculate a hash of all custom units
129 hash_md5 = md5()
131 try:
132 from common.models import CustomUnit
134 custom_units = list(CustomUnit.objects.all())
135 except Exception:
136 # Database is likely not ready
137 custom_units = []
139 for cu in custom_units: 139 ↛ 140line 139 didn't jump to line 140 because the loop on line 139 never started
140 try:
141 fmt = cu.fmt_string()
142 reg.define(fmt)
144 hash_md5.update(fmt.encode('utf-8'))
146 except Exception as e:
147 logger.exception('Failed to load custom unit: %s - %s', cu.fmt_string(), e)
149 # Once custom units are loaded, save registry
150 _unit_registry = reg
152 # Update the unit registry hash
153 set_unit_registry_hash(hash_md5.hexdigest())
155 dt = time.time() - t_start
156 logger.debug('Loaded unit registry in %.3f s', dt)
158 return reg
161def from_engineering_notation(value):
162 """Convert a provided value to 'natural' representation from 'engineering' notation.
164 Ref: https://en.wikipedia.org/wiki/Engineering_notation
166 In "engineering notation", the unit (or SI prefix) is often combined with the value,
167 and replaces the decimal point.
169 Examples:
170 - 1K2 -> 1.2K
171 - 3n05 -> 3.05n
172 - 8R6 -> 8.6R
174 And, we should also take into account any provided trailing strings:
176 - 1K2 ohm -> 1.2K ohm
177 - 10n005F -> 10.005nF
178 """
179 value = str(value).strip()
181 pattern = r'(\d+)([a-zA-Z]+)(\d+)(.*)'
183 if match := re.match(pattern, value):
184 left, prefix, right, suffix = match.groups()
185 return f'{left}.{right}{prefix}{suffix}'
187 return value
190def convert_value(value, unit=None):
191 """Attempt to convert a value to a specified unit.
193 Arguments:
194 value: The value to convert
195 unit: The target unit to convert to
197 Returns:
198 The converted value (ideally a pint.Quantity value)
200 Raises:
201 Exception if the value cannot be converted to the specified unit
202 """
203 ureg = get_unit_registry()
205 # Convert the provided value to a pint.Quantity object
206 value = ureg.Quantity(value)
208 # Convert to the specified unit
209 if unit:
210 if is_dimensionless(value):
211 magnitude = value.to_base_units().magnitude
212 value = ureg.Quantity(magnitude, unit)
213 else:
214 value = value.to(unit)
216 return value
219def convert_physical_value(value: str, unit: Optional[str] = None, strip_units=True):
220 """Validate that the provided value is a valid physical quantity.
222 Arguments:
223 value: Value to validate (str)
224 unit: Optional unit to convert to, and validate against
225 strip_units: If True, strip units from the returned value, and return only the dimension
227 Returns:
228 The converted quantity, in the specified units
230 Raises:
231 ValidationError: If the value is invalid or cannot be converted to the specified unit
232 """
233 ureg = get_unit_registry()
235 # Check that the provided unit is available in the unit registry
236 if unit:
237 try:
238 valid = unit in ureg
239 except Exception:
240 valid = False
242 if not valid:
243 raise ValidationError(_(f'Invalid unit provided ({unit})'))
245 original = str(value).strip()
247 # Ensure that the value is a string
248 value = str(value).strip() if value else ''
249 unit = str(unit).strip() if unit else ''
251 # Handle imperial length measurements
252 if value.count("'") == 1 and value.endswith("'"):
253 value = value.replace("'", ' feet')
255 if value.count('"') == 1 and value.endswith('"'):
256 value = value.replace('"', ' inches')
258 # Error on blank values
259 if not value:
260 raise ValidationError(_('No value provided'))
262 # Construct a list of values to "attempt" to convert
263 attempts = [value]
265 # Attempt to convert from engineering notation
266 eng = from_engineering_notation(value)
267 attempts.append(eng)
269 # Append the unit, if provided
270 # These are the "final" attempts to convert the value, and *must* appear after previous attempts
271 if unit:
272 attempts.append(f'{value}{unit}')
273 attempts.append(f'{eng}{unit}')
275 value: Optional[str] = None
277 # Run through the available "attempts", take the first successful result
278 for attempt in attempts:
279 try:
280 value = convert_value(attempt, unit)
281 break
282 except Exception:
283 value = None
285 if value is None:
286 if unit:
287 raise ValidationError(_(f'Could not convert {original} to {unit}'))
288 else:
289 raise ValidationError(_('Invalid quantity provided'))
291 # Calculate the "magnitude" of the value, as a float
292 # If the value is specified strangely (e.g. as a fraction or a dozen), this can cause issues
293 # So, we ensure that it is converted to a floating point value
294 # If we wish to return a "raw" value, some trickery is required
295 try:
296 if unit:
297 magnitude = ureg.Quantity(value.to(ureg.Unit(unit))).magnitude
298 else:
299 magnitude = ureg.Quantity(value.to_base_units()).magnitude
301 magnitude = float(ureg.Quantity(magnitude).to_base_units().magnitude)
302 except Exception as exc:
303 raise ValidationError(_('Invalid quantity provided') + f': ({exc})')
305 if strip_units:
306 return magnitude
307 elif unit or value.units:
308 return ureg.Quantity(magnitude, unit or value.units)
309 return ureg.Quantity(magnitude)
312def is_dimensionless(value):
313 """Determine if the provided value is 'dimensionless'.
315 A dimensionless value might look like:
317 0.1
318 1/2 dozen
319 three thousand
320 1.2 dozen
321 (etc)
322 """
323 ureg = get_unit_registry()
325 # Ensure the provided value is in the right format
326 value = ureg.Quantity(value)
328 if value.units == ureg.dimensionless:
329 return True
331 return value.to_base_units().units == ureg.dimensionless