|
|
|
@ -10,8 +10,8 @@ from collections import namedtuple
@@ -10,8 +10,8 @@ from collections import namedtuple
|
|
|
|
|
from enum import Enum |
|
|
|
|
from io import StringIO |
|
|
|
|
|
|
|
|
|
DROP_CHECK_SUFFIX = ".cor" |
|
|
|
|
ENVIRONMENT_SUFFIX = "_Logfile.log" |
|
|
|
|
|
|
|
|
|
PrintLogResult = namedtuple("PrintLogResult", ["environment", "info"]) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class CheckWhen(Enum): |
|
|
|
@ -25,7 +25,7 @@ class CheckResult(Enum):
@@ -25,7 +25,7 @@ class CheckResult(Enum):
|
|
|
|
|
SKIPPED = "skipped" |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class LogResult: |
|
|
|
|
class DropCheckResult: |
|
|
|
|
def __init__( |
|
|
|
|
self, |
|
|
|
|
path, |
|
|
|
@ -101,18 +101,13 @@ def parse_log_line(line, cast_to, default_value=np.nan, separator="="):
@@ -101,18 +101,13 @@ def parse_log_line(line, cast_to, default_value=np.nan, separator="="):
|
|
|
|
|
return parse_str_value(str_data, cast_to, default_value) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def get_log_files(folder, suffix=".cor"): |
|
|
|
|
visible = (p for p in folder.iterdir() if not p.name.startswith(".")) |
|
|
|
|
return [p for p in visible if p.name.endswith(suffix)] |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def parse_log_files(log_list): |
|
|
|
|
pre_run = dict() |
|
|
|
|
post_run = dict() |
|
|
|
|
well_list = list() |
|
|
|
|
# use the files sorted by date and time |
|
|
|
|
for path in sorted(log_list): |
|
|
|
|
log_result = LogResult.from_file(path) |
|
|
|
|
log_result = DropCheckResult.from_file(path) |
|
|
|
|
if log_result.well not in pre_run: |
|
|
|
|
log_result.when = CheckWhen.PRE_RUN |
|
|
|
|
pre_run[log_result.well] = log_result |
|
|
|
@ -125,7 +120,9 @@ def parse_log_files(log_list):
@@ -125,7 +120,9 @@ def parse_log_files(log_list):
|
|
|
|
|
|
|
|
|
|
skipped_runs = {well for well in pre_run if well not in post_run} |
|
|
|
|
for well in skipped_runs: |
|
|
|
|
post_result = LogResult("", well, CheckResult.SKIPPED, when=CheckWhen.POST_RUN) |
|
|
|
|
post_result = DropCheckResult( |
|
|
|
|
"", well, CheckResult.SKIPPED, when=CheckWhen.POST_RUN |
|
|
|
|
) |
|
|
|
|
post_run[well] = post_result |
|
|
|
|
|
|
|
|
|
parsed_files = [] |
|
|
|
@ -136,12 +133,53 @@ def parse_log_files(log_list):
@@ -136,12 +133,53 @@ def parse_log_files(log_list):
|
|
|
|
|
return pd.DataFrame([pf.as_dict() for pf in parsed_files]) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def parse_environment_log(log_file_path): |
|
|
|
|
with open(log_file_path, "r", encoding="iso-8859-1") as file_handle: |
|
|
|
|
env_lines = [l for l in file_handle if "\tHumidity=\t" in l] |
|
|
|
|
def split_print_log_line(line): |
|
|
|
|
_, value = line.split(":", 1) |
|
|
|
|
return value.strip() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def count_solutions(file_handle): |
|
|
|
|
solutions = set() |
|
|
|
|
for line in file_handle: |
|
|
|
|
line = line.strip() |
|
|
|
|
if not line or line[0] in ("X", "Y", "F", "["): |
|
|
|
|
# empty line or uninteresting one, pick next one |
|
|
|
|
continue |
|
|
|
|
elif line.startswith("Drops/Field"): |
|
|
|
|
# finished with all field definition, leave loop |
|
|
|
|
break |
|
|
|
|
entries = (item.strip() for item in line.split("\t")) |
|
|
|
|
wells = (well for well in entries if well) |
|
|
|
|
solutions.update(wells) |
|
|
|
|
return len(solutions) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def parse_print_log(log_files): |
|
|
|
|
env_lines = [] |
|
|
|
|
print_info = {} |
|
|
|
|
with open(log_files, "r", encoding="iso-8859-1") as file_handle: |
|
|
|
|
for line in file_handle: |
|
|
|
|
if "\tHumidity=\t" in line: |
|
|
|
|
env_lines.append(line) |
|
|
|
|
elif line.startswith("Probe:"): |
|
|
|
|
print_info["source"] = split_print_log_line(line) |
|
|
|
|
elif line.startswith("Target:"): |
|
|
|
|
target_and_fields = split_print_log_line(line) |
|
|
|
|
target, fields = target_and_fields.rsplit(":", 1) |
|
|
|
|
print_info["target"] = target.strip() |
|
|
|
|
print_info["fields"] = len(fields.split(",")) |
|
|
|
|
elif line.startswith("Humidity:"): |
|
|
|
|
print_info["humidity"] = split_print_log_line(line) |
|
|
|
|
elif line.startswith("Run Name:"): |
|
|
|
|
print_info["run"] = split_print_log_line(line) |
|
|
|
|
elif line.startswith("Dot Pitch:"): |
|
|
|
|
# important to pass the filehandle iterator here |
|
|
|
|
print_info["solutions"] = count_solutions(file_handle) |
|
|
|
|
|
|
|
|
|
buff = StringIO("".join(env_lines)) |
|
|
|
|
columns = ["datetime", "garbage 1", "humidity", "garbage 2", "temperature"] |
|
|
|
|
df = pd.read_csv( |
|
|
|
|
tmp_df = pd.read_csv( |
|
|
|
|
buff, sep="\t", header=None, names=columns, index_col=0, parse_dates=True |
|
|
|
|
) |
|
|
|
|
return df.drop(columns=["garbage 1", "garbage 2"]) |
|
|
|
|
environment_df = tmp_df.drop(columns=["garbage 1", "garbage 2"]) |
|
|
|
|
return PrintLogResult(environment_df, print_info) |
|
|
|
|