Spaces:
Running
Running
File size: 7,262 Bytes
0d3476b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 |
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
@app.route('/lab_work', methods=['POST'])
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
|