-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
5 changed files
with
80 additions
and
24 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,30 +1,54 @@ | ||
import string | ||
from numbers import Number | ||
from typing import Any, Optional | ||
from typing import Optional | ||
|
||
import numpy as np | ||
from loguru import logger | ||
|
||
|
||
def validate_cas(s: Any) -> Optional[str]: | ||
ERROR = "CAS Check Digit error: CAS '{}' has check digit of {}, but it should be {}" | ||
def calculate_check_digit(cas: str) -> int: | ||
return sum((a + 1) * int(b) for a, b in zip(range(9), cas[-1::-1])) % 10 | ||
|
||
if isinstance(s, str): | ||
s = s.strip() | ||
if not s: | ||
|
||
def validate_cas_string(cas: Optional[str]) -> Optional[str]: | ||
if isinstance(cas, str): | ||
cas = cas.strip() | ||
if not cas: | ||
return None | ||
elif isinstance(s, Number) and np.isnan(s): | ||
elif isinstance(cas, Number) and np.isnan(cas): | ||
return None | ||
|
||
if s[-1] not in string.digits: | ||
check_digit = None | ||
cas = s.replace("-", "") | ||
if "-" not in cas: | ||
first, second, check_digit = cas[:-3], cas[-3:-1], int(cas[-1]) | ||
if str(calculate_check_digit(first + second)) != str(check_digit): | ||
logger.warning( | ||
"Removing invalid CAS number {}; last digit should be {}".format( | ||
cas, check_digit | ||
) | ||
) | ||
return None | ||
return "-".join([first, second, str(check_digit)]).lstrip("0") | ||
elif cas.count("-") == 2 and not cas.split("-")[2]: | ||
# e.g. 1228284-64- | ||
check_digit = str(calculate_check_digit(cas.replace("-", ""))) | ||
logger.warning( | ||
"Adding missing CAS check digit, {} -> {}".format(cas, cas + check_digit) | ||
) | ||
return cas + check_digit | ||
elif cas.count("-") == 2: | ||
first, second, third = cas.split("-") | ||
check_digit = calculate_check_digit(first + second) | ||
if str(check_digit) != third: | ||
logger.warning( | ||
"Removing invalid CAS number {}; last digit should be {}".format( | ||
cas, check_digit | ||
) | ||
) | ||
else: | ||
return cas.lstrip("0") | ||
else: | ||
check_digit = s[-1] | ||
cas = s[:-1].replace('-', '') | ||
|
||
total = sum((a + 1) * int(b) for a, b in zip(range(9), s.replace("-", "")[-2::-1])) | ||
if not total % 10 == int(s[-1]): | ||
logger.warning("CAS not valid: {} ({})".format(s, ERROR.format(s, s[-1], total % 10))) | ||
logger.warning( | ||
"Given CAS can't be validated, wrong number of hyphens are present: {}".format( | ||
cas | ||
) | ||
) | ||
return None | ||
return s.lstrip("0") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,32 @@ | ||
from bw_simapro_csv import SimaProCSV | ||
from bw_simapro_csv.blocks import GenericBiosphere | ||
from bw_simapro_csv.cas import calculate_check_digit, validate_cas_string | ||
|
||
|
||
def test_calculate_check_digit(): | ||
assert calculate_check_digit("773218") == 5 | ||
assert calculate_check_digit("778240") == 3 | ||
|
||
|
||
def test_validate_cas_string(): | ||
assert validate_cas_string("7782425") == "7782-42-5" | ||
assert validate_cas_string("007782425") == "7782-42-5" | ||
assert validate_cas_string(" 7782-42-5\n") == "7782-42-5" | ||
assert validate_cas_string("007782-42-5") == "7782-42-5" | ||
assert validate_cas_string("1228284-64-") == "1228284-64-7" | ||
assert validate_cas_string("") is None | ||
assert validate_cas_string(None) is None | ||
assert validate_cas_string(float("NaN")) is None | ||
assert validate_cas_string("7782-425") is None | ||
assert validate_cas_string("7782424") is None | ||
|
||
|
||
def test_cas_in_file(fixtures_dir): | ||
obj = SimaProCSV(fixtures_dir / "cas_missing_check_number.csv") | ||
blocks = [ | ||
elem | ||
for elem in obj.blocks | ||
if isinstance(elem, GenericBiosphere) and elem.category == "Emissions to soil" | ||
] | ||
expected = [None, "1228284-64-7"] | ||
assert [obj["cas_number"] for obj in blocks[0].parsed] == expected |