Spaces:
Running
Running
import json | |
import logging | |
import re | |
from flask import Flask, request, jsonify | |
from routes import app | |
logging.basicConfig(level=logging.INFO) | |
logger = logging.getLogger(__name__) | |
def parse_markdown_table(table_str): | |
""" | |
Parses a markdown table string and returns a list of lab configurations. | |
Each configuration is a dictionary with keys: | |
- lab: int | |
- cell_counts: list of ints | |
- increment: str | |
- condition: list of three ints | |
""" | |
lines = table_str.strip().split('\n') | |
# Remove header and separator | |
data_lines = lines[2:] | |
lab_configs = {} | |
for line in data_lines: | |
# Split by '|' and strip spaces | |
parts = [part.strip() for part in line.strip().split('|')] | |
if len(parts) < 4: | |
continue # Skip incomplete lines | |
lab_str, cell_counts_str, increment_str, condition_str = parts[:4] | |
lab = int(lab_str) | |
# Extract cell counts | |
cell_counts = list(map(int, cell_counts_str.split())) | |
# Extract increment expression | |
increment = increment_str | |
# Extract condition | |
# Use regular expression to extract numbers, considering multiple spaces | |
condition_numbers = re.findall(r'\d+', condition_str) | |
condition = list(map(int, condition_numbers)) | |
if len(condition) != 3: | |
raise ValueError("Condition does not have exactly three numbers.") | |
# Store in lab_configs | |
lab_configs[lab] = { | |
'cell_counts': cell_counts, | |
'increment': increment, | |
'condition': condition | |
} | |
return lab_configs | |
def apply_increment(count, expr): | |
""" | |
Applies the increment expression to the given count. | |
""" | |
tokens = expr.split() | |
if len(tokens) != 3: | |
raise ValueError(f"Invalid increment expression: {expr}") | |
op1, operator, op2 = tokens | |
if op1 != 'count': | |
raise ValueError(f"Invalid operand: {op1}") | |
if op2 == 'count': | |
operand = count | |
else: | |
operand = int(op2) | |
if operator == '+': | |
return count + operand | |
elif operator == '*': | |
return count * operand | |
else: | |
raise ValueError(f"Unknown operator: {operator}") | |
def evaluate_condition(count, condition): | |
""" | |
Evaluates the condition and returns the target lab. | |
Condition is a list of three integers: [divisor, lab_if_true, lab_if_false] | |
""" | |
divisor, lab_if_true, lab_if_false = condition | |
if divisor == 0: | |
raise ValueError("Divisor in condition cannot be zero.") | |
if count % divisor == 0: | |
return lab_if_true | |
else: | |
return lab_if_false | |
def lab_work(): | |
try: | |
data = request.get_json() | |
if not isinstance(data, list): | |
return jsonify({"error": "Input data should be a list of markdown tables."}), 400 | |
test_cases = data # List of markdown table strings | |
output = [] | |
for idx, table_str in enumerate(test_cases): | |
logger.info(f"Processing Test Case {idx + 1}") | |
# Parse the table | |
lab_configs = parse_markdown_table(table_str) | |
if not lab_configs: | |
return jsonify({"error": f"No lab configurations found in Test Case {idx + 1}."}), 400 | |
# Determine the number of labs | |
lab_numbers = sorted(lab_configs.keys()) | |
num_labs = len(lab_numbers) | |
max_lab_number = max(lab_numbers) | |
# Initialize labs data | |
labs = {} | |
for lab in lab_numbers: | |
labs[lab] = { | |
'increment': lab_configs[lab]['increment'], | |
'condition': lab_configs[lab]['condition'], | |
'current_dishes': lab_configs[lab]['cell_counts'][:], # Make a copy | |
'next_day_dishes': [] | |
} | |
# Initialize analysis counts | |
analysis_counts = {lab: 0 for lab in lab_numbers} | |
# Initialize records at intervals | |
records = {} | |
# Simulate 10,000 days | |
for day in range(1, 10001): | |
# Initialize processing lists for the day | |
processing_lists = {lab: labs[lab]['current_dishes'][:] for lab in lab_numbers} | |
# Clear current_dishes for receiving new dishes passed to labs already processed | |
for lab in lab_numbers: | |
labs[lab]['current_dishes'] = [] | |
# Process labs in order | |
for lab in lab_numbers: | |
dishes_to_process = processing_lists[lab] | |
for dish in dishes_to_process: | |
# Apply increment | |
try: | |
updated_count = apply_increment(dish, labs[lab]['increment']) | |
except ValueError as ve: | |
return jsonify({"error": f"Error in increment expression for lab {lab}: {str(ve)}"}), 400 | |
# Evaluate condition | |
try: | |
target_lab = evaluate_condition(updated_count, labs[lab]['condition']) | |
except ValueError as ve: | |
return jsonify({"error": f"Error in condition for lab {lab}: {str(ve)}"}), 400 | |
# Increment analysis count | |
analysis_counts[lab] += 1 | |
# Assign to target lab | |
if target_lab not in lab_configs: | |
return jsonify({"error": f"Target lab {target_lab} does not exist in Test Case {idx + 1}."}), 400 | |
if target_lab > lab: | |
# Target lab has not been processed yet today | |
processing_lists[target_lab].append(updated_count) | |
else: | |
# Target lab has already been processed today | |
labs[target_lab]['next_day_dishes'].append(updated_count) | |
# After processing all labs, update current_dishes for next day | |
for lab in lab_numbers: | |
# Append dishes passed to labs already processed today | |
labs[lab]['current_dishes'].extend(labs[lab]['next_day_dishes']) | |
# Clear next_day_dishes for the next day | |
labs[lab]['next_day_dishes'] = [] | |
# Record counts at intervals of 1000 days | |
if day % 1000 == 0: | |
# Prepare the counts list ordered by lab numbers | |
counts_list = [analysis_counts[lab] for lab in lab_numbers] | |
records[str(day)] = counts_list | |
logger.info(f"Recorded counts for day {day}: {counts_list}") | |
# After simulation, prepare the output for this test case | |
output.append(records) | |
return jsonify(output), 200 | |
except Exception as e: | |
logger.error(f"An error occurred: {str(e)}") | |
return jsonify({"error": f"An error occurred: {str(e)}"}), 500 | |