main.py
initializeSystem
async def initializeSystem():
network = ipaddress.ip_network('192.168.1.0/24')
modules = []
async with aiohttp.ClientSession() as session:
tasks = []
for ip in network.hosts():
tasks.append(asyncio.create_task(initialize_host(ip, session, modules)))
await asyncio.gather(*tasks)
return modules
The initializeSystem method is responsible for setting up the fireworks control system at the start of the application. It initializes the system, creates the instances for the Module class, adds hosts to the modules, and sets up the ignition points. This method is the primary setup procedure that enables the other parts of the system to function properly.
initialize_host
async def initialize_host(ip, session, modules):
try:
async with session.post(f'http://{ip}/initialize', headers=headers, timeout=3) as response:
if response.status == 200:
data = await response.json()
await process_response(ip, data, modules)
except Exception:
pass
The initialize_host method is used to instantiate a Host class object for each host specified in the hosts list. It sets the host's ipAddress, modules, moduleId, moduleName, and ignitions. This method is integral to the setup of the system as it defines the hosts that the modules will be interacting with.
process_response
async def process_response(ip, data, modules):
moduleName = data['moduleName']
moduleId = data['moduleId']
ignitions = [Ignition(f'http://{ip}/ignite/{i+1}') for i in range(data['ignitions'])]
existing_module = find_module_by_name(modules, moduleName)
if existing_module is not None:
existing_module.add_host(Host(ip, data['modules'], moduleId, moduleName, ignitions))
else:
module = Module(moduleName)
module.add_host(Host(ip, data['modules'], moduleId, moduleName, ignitions))
modules.append(module)
The process_response method handles the responses received from the user command input. Depending on the command type, it performs various operations such as firing a specified ignition, generating a sequence of ignitions, swapping the order of ignitions in the sequence, etc. This method is a key part of the command-and-control aspect of the application, ensuring that user commands are correctly processed and executed.
find_module_by_name
def find_module_by_name(modules, moduleName):
for module in modules:
if module.moduleName == moduleName:
return module
return None
The find_module_by_name method is a helper function used to find and return a module object given its name. It iterates over the list of modules in the system and checks if the name of each module matches the provided name. If a match is found, it returns the corresponding module object. This method simplifies the process of accessing module-specific data and operations within the system.
generate_random_sequence
def generate_random_sequence(modules):
ignitions = []
# Iterate over each module
for module in modules:
# Iterate over each host in the module
for host in module.hosts:
# Iterate over each ignition in the host
for ignition in host.ignitions:
if not ignition.hasBeenFired:
ignitions.append(ignition)
# Shuffle the list of ignitions
random.shuffle(ignitions)
# Return the shuffled list
return ignitions
The generate_random_sequence method is responsible for generating a random sequence of ignition orders. This sequence is represented as a list of numbers, each number representing the order in which an ignition will be fired. By providing a randomized sequence, this method introduces an element of unpredictability into the system, which can be useful in a variety of scenarios, such as creating a random firework display.
get_order_from_url
def get_order_from_url(url):
path = urlparse(url).path
return int(path.split('/')[-1])
The get_order_from_url method is used to extract and return the order value from a given URL. This method is essential for managing the order of ignitions as they are specified in their endpoint URLs. By extracting this information, the method enables the system to properly arrange and manage the sequence of ignitions.
generate_sequential_sequence
def generate_sequential_sequence(modules):
# Step 1: Create a grouped list of all ignitions
ignition_groups = defaultdict(lambda: defaultdict(deque))
for module in modules:
for host in module.hosts:
for ignition in host.ignitions:
order = get_order_from_url(ignition.endpoint)
key = f"{host.moduleName}-{host.ipAddress}"
ignition_groups[order][key].append(ignition)
# Step 2: Iterate over the groups in sequential order and within each group,
# cycle through the subgroups
ignition_sequence = []
last_module = None
for order in sorted(ignition_groups.keys()):
subgroups = list(ignition_groups[order].items())
while subgroups:
# Get the subgroup key (moduleName-ipAddress) and its ignitions
key, ignitions = subgroups.pop(0)
module_name = key.split('-')[0]
# If this module is the same as the last module, and there are other subgroups,
# put it at the end of the list and continue with the next one
if module_name == last_module and subgroups:
subgroups.append((key, ignitions))
continue
# Add the next ignition from this subgroup to the sequence
ignition_sequence.append(ignitions.popleft())
# If there are more ignitions in this subgroup, put it back in the list
if ignitions:
subgroups.append((key, ignitions))
last_module = module_name
return ignition_sequence
The generate_sequential_sequence method is used to generate a sequential sequence of ignition orders. This sequence is represented as a list of numbers, each number representing the order in which an ignition will be fired. This method ensures that the sequence is always generated in a consistent and predictable manner, starting from 1 and incrementing by 1 for each subsequent ignition. This can be especially useful in scenarios where a specific order of ignitions needs to be maintained, such as creating a sequential firework display.
sequence_fire
async def sequence_fire(current_sequence, modules):
for sequence_ignition in current_sequence:
for module in modules:
for host in module.hosts:
for ignition in host.ignitions:
if ignition.endpoint == sequence_ignition.endpoint:
await fire(ignition)
print("\n" * 100)
for module in modules:
for host in module.hosts:
PrintUtils.sequence(display_sequence(host, current_sequence))
print("-------")
print("\n" * 100)
The sequence_fire method is responsible for triggering the ignition of fireworks in a specific sequence. It takes a sequence list as an input, where each element represents the order of the ignitions. The method iterates over the sequence and fires each ignition in the order specified by the sequence. The sequence_fire method provides a way to create a more organized and controlled firework display, by allowing for a precise order of ignitions to be executed.
fire
async def fire(ignition):
if not ignition.hasBeenFired:
await ignition.fire()
PrintUtils.log(f"Ignition fired successfully.")
else:
PrintUtils.error(f"Ignition already received.")
The fire method is used to trigger the ignition of a specific firework module. It takes two parameters: the module name and the ignition index. The method first finds the module by name using the find_module_by_name method. If the module is found, the specified ignition is fired. If either the module or the ignition index is not found, the method will print an error message. This method provides a direct way to trigger a single firework ignition, allowing for specific control over which firework is ignited at a given time.
find_and_fire
async def find_and_fire(modules, moduleName, moduleId, igniteNumber):
# Find the module that matches the moduleName
for module in modules:
if module.moduleName == moduleName:
# Find the host with the matching moduleId
for host in module.hosts:
if host.moduleId == moduleId:
# Find the Ignition with the matching igniteNumber
for ignition in host.ignitions:
if int(ignition.endpoint.split('/')[-1]) == igniteNumber:
await fire(ignition)
return
PrintUtils.error(f"Ignition with igniteNumber '{igniteNumber}' not found in module '{moduleName}', host '{moduleId}'.")
return
PrintUtils.error(f"Host with moduleId '{moduleId}' not found in module '{moduleName}'.")
return
PrintUtils.error(f"Module '{moduleName}' not found.")
The find_and_fire method combines the functionality of finding a specific module and firing an ignition. It takes three parameters: the name of the module, the module ID, and the ignition index. It first retrieves the module using the find_module_by_name method, and if successful, it proceeds to fire the specified ignition within that module using the fire method. If either the module or the ignition is not found, an error message is displayed. This method offers a convenient way to find and fire a specific ignition within a specified module.
display_sequence
def display_sequence(host, ignition_sequence):
# Get the IP and module details
result = f"{host.ipAddress} - {host.moduleName} {host.moduleId}/{host.modules} - Ignitions: "
# Generate the ignition list
for ignition in host.ignitions:
if ignition.hasBeenFired:
result += f"[{chr(0x25CF)}] "
elif ignition in ignition_sequence:
result += f"[{ignition_sequence.index(ignition) + 1}] "
return result
The display_sequence method is responsible for visually representing the sequence of ignitions. It takes a sequence list as input, where each element represents the order of the ignitions. The method iterates over the sequence and prints out a representation of each ignition, using the format specified in the PrintUtils.sequence method. This allows for a clear and easy-to-understand visualization of the sequence of ignitions that is set to occur.
sequence_swap
def sequence_swap(sequence, index1, index2):
# Check if the indexes are within the list
if index1 <= len(sequence) and index2 <= len(sequence):
# Subtract 1 from each index to make them 0-based
index1 -= 1
index2 -= 1
# Swap the elements
sequence[index1], sequence[index2] = sequence[index2], sequence[index1]
else:
PrintUtils.error(f"Indices are out of range. List length: {len(sequence)}, Received Indices: {index1}, {index2}")
return sequence
The sequence_swap method is used to change the order of ignitions in a sequence. It takes two parameters: index1 and index2, which represent the indices of the ignitions in the sequence that are to be swapped. The method checks whether the provided indices are within the range of the sequence. If they are, it swaps the ignitions at these indices. If they are not, it prints an error message indicating the indices are out of range. This method provides a way to alter the sequence of ignitions after it has been generated, allowing for greater flexibility in the execution of the ignitions.
main
async def main():
modules = []
current_sequence = []
while True:
command = input('Enter a command: ')
if command.lower() == 'init':
if not modules:
modules = await initializeSystem()
for module in modules:
for host in module.hosts:
PrintUtils.host(host.display())
print("-------")
elif command.lower() == 'fire sequence':
if not current_sequence:
PrintUtils.param_error(f'Sequence generation required. Please use "generate sequence -f <format>".')
else:
await sequence_fire(current_sequence, modules)
for module in modules:
for host in module.hosts:
PrintUtils.host(host.display())
print("-------")
elif command.lower().startswith('fire'):
fire_command = command[5:].strip()
fire_params = ParseUtils.parse_fire_command(fire_command)
if not fire_params:
PrintUtils.param_error(f'Invalid command format. Please use "fire -m <moduleName> -d <moduleId> -i <igniteNumber>".')
continue
moduleName, moduleId, igniteNumber = fire_params
PrintUtils.log(f"Fire command received for Module: {moduleName}, Module ID: {moduleId}, Ignite Number: {igniteNumber}")
await find_and_fire(modules, moduleName, moduleId, igniteNumber)
for module in modules:
for host in module.hosts:
PrintUtils.host(host.display())
print("-------")
elif command.lower().startswith('generate sequence'):
generate_sequence_params = ParseUtils.parse_generate_sequence(command)
if not generate_sequence_params:
PrintUtils.param_error(f'Invalid command format. Please use "generate sequence -f <format>".')
continue
command, modifier, seq_format = generate_sequence_params
if not current_sequence:
PrintUtils.log(f"Generate Sequence command received for {seq_format}: Generating Sequence now.")
if seq_format == "RANDOM":
current_sequence = generate_random_sequence(modules)
elif seq_format == "SEQUENTIAL":
current_sequence = generate_sequential_sequence(modules)
for module in modules:
for host in module.hosts:
PrintUtils.sequence(display_sequence(host, current_sequence))
print("-------")
elif command.lower().startswith('sequence swap'):
swap_params = ParseUtils.parse_sequence_swap(command)
if swap_params:
index1, index2 = swap_params
current_sequence = sequence_swap(current_sequence, index1, index2)
else:
PrintUtils.param_error("Invalid command format. Please use 'sequence swap <index1> <index2>'.")
for module in modules:
for host in module.hosts:
PrintUtils.sequence(display_sequence(host, current_sequence))
print("-------")
elif command.lower() == 'override sequence':
current_sequence = []
PrintUtils.log(f"Current Sequence reset")
elif command.lower() == 'override reboot':
modules = []
PrintUtils.log(f"System reset")
else:
PrintUtils.error('Invalid command. Try again')
The main function is the entry point of the script. It begins by loading modules (firework units) from a JSON file, converting them into Module objects for easy manipulation. Then, it enters an infinite loop to keep the program running and listening for user input.
The function uses Python's input function to accept commands from the user, which are then parsed and executed. The command exit can be used to terminate the loop and end the program.
The commands supported are:
fire- This command fires a specific ignition of a module.display- This command displays the current state of the modules, hosts, and ignitions in the network.generate sequence- This command generates a firing sequence either randomly or sequentially.sequence swap- This command swaps two ignitions in the firing sequence.
In case of an error or invalid command, the function handles it gracefully and prints an error message.
Overall, the main function coordinates the user interactions, parsing and executing commands, displaying the state, and managing the sequence of ignition.