178 lines
5.0 KiB
Python
Executable File
178 lines
5.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Имитатор параллельной шины для gpio-interrupt (Raspberry Pi 4)
|
|
"""
|
|
|
|
import gpiod
|
|
import time
|
|
import argparse
|
|
import sys
|
|
import signal
|
|
from gpiod.line import Direction, Value
|
|
|
|
|
|
class ParallelBusImitator:
|
|
"""Имитатор 7-битной параллельной шины данных"""
|
|
|
|
# Временные константы (секунды)
|
|
DATA_SETUP_TIME = 0.000005
|
|
STROBE_WIDTH = 0.000005
|
|
HOLD_TIME = 0.000005
|
|
|
|
WR_PIN = 12 # BCM12
|
|
|
|
DATA_PINS_MAPPING = [
|
|
(21, 18, "D0 (LSB)"),
|
|
(7, 27, "D1"),
|
|
(6, 22, "D2"),
|
|
(5, 23, "D3"),
|
|
(25, 17, "D4"),
|
|
(24, 20, "D5"),
|
|
(23, 21, "D6"),
|
|
(22, None, "D7 (не подключён)"),
|
|
]
|
|
|
|
def __init__(self, use_mask=True, chip_path='/dev/gpiochip0'):
|
|
self.use_mask = use_mask
|
|
self.counter = 0
|
|
self.active_pins = []
|
|
|
|
for wpi, bcm, desc in self.DATA_PINS_MAPPING:
|
|
if bcm is not None:
|
|
self.active_pins.append((wpi, bcm, desc))
|
|
|
|
self.data_pins_bcm = [bcm for _, bcm, _ in self.active_pins]
|
|
self.all_pins = self.data_pins_bcm + [self.WR_PIN]
|
|
|
|
try:
|
|
self.chip = gpiod.Chip(chip_path)
|
|
self.lines = self.chip.request_lines(
|
|
{
|
|
p: gpiod.LineSettings(
|
|
direction=Direction.OUTPUT,
|
|
output_value=Value.INACTIVE
|
|
)
|
|
for p in self.all_pins
|
|
},
|
|
consumer="par-bus-imitator"
|
|
)
|
|
except PermissionError:
|
|
print("Нет прав доступа к GPIO. Используйте sudo.")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"Ошибка GPIO: {e}")
|
|
sys.exit(1)
|
|
|
|
bit_width = 4 if use_mask else len(self.data_pins_bcm)
|
|
|
|
print(f"✓ Имитатор шины ({bit_width}-бит режим)")
|
|
print("Нажмите Ctrl+C для остановки\n")
|
|
|
|
def send_byte(self, value):
|
|
"""Отправка одного байта"""
|
|
if self.use_mask:
|
|
value &= 0x0F
|
|
else:
|
|
value &= 0x7F
|
|
|
|
out = {}
|
|
|
|
for bit_pos, bcm_pin in enumerate(self.data_pins_bcm):
|
|
bit = (value >> bit_pos) & 1
|
|
out[bcm_pin] = Value.ACTIVE if bit else Value.INACTIVE
|
|
|
|
self.lines.set_values(out)
|
|
time.sleep(self.DATA_SETUP_TIME)
|
|
|
|
self.lines.set_value(self.WR_PIN, Value.ACTIVE)
|
|
time.sleep(self.STROBE_WIDTH)
|
|
self.lines.set_value(self.WR_PIN, Value.INACTIVE)
|
|
time.sleep(self.HOLD_TIME)
|
|
|
|
self.counter += 1
|
|
self.show_status(value)
|
|
|
|
def show_status(self, value):
|
|
"""Красивый статус без спама"""
|
|
bits = 4 if self.use_mask else 7
|
|
bin_val = format(value, f'0{bits}b')
|
|
|
|
msg = (
|
|
f"\r▶ Передача: "
|
|
f"HEX=0x{value:02X} | "
|
|
f"BIN={bin_val} | "
|
|
f"Пакетов: {self.counter}"
|
|
)
|
|
|
|
print(msg, end='', flush=True)
|
|
|
|
def send_sequence(self, values, delay=0.2):
|
|
for val in values:
|
|
self.send_byte(val)
|
|
time.sleep(delay)
|
|
|
|
def cleanup(self):
|
|
if hasattr(self, 'lines'):
|
|
self.lines.release()
|
|
if hasattr(self, 'chip'):
|
|
self.chip.close()
|
|
|
|
|
|
def handle_signal(signum, frame):
|
|
print("\nЗавершение работы...")
|
|
sys.exit(0)
|
|
|
|
|
|
def main():
|
|
signal.signal(signal.SIGINT, handle_signal)
|
|
signal.signal(signal.SIGTERM, handle_signal)
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description='Имитатор параллельной шины'
|
|
)
|
|
|
|
parser.add_argument('--no-mask', action='store_true',
|
|
help='7-битный режим')
|
|
parser.add_argument('--test',
|
|
choices=['default', 'counter', 'pattern'],
|
|
default='default')
|
|
parser.add_argument('--delay', type=float, default=0.2)
|
|
parser.add_argument('--single', type=str)
|
|
parser.add_argument('--chip', default='/dev/gpiochip0')
|
|
|
|
args = parser.parse_args()
|
|
|
|
imitator = ParallelBusImitator(
|
|
use_mask=not args.no_mask,
|
|
chip_path=args.chip
|
|
)
|
|
|
|
try:
|
|
if args.single:
|
|
val = int(args.single, 16)
|
|
imitator.send_byte(val)
|
|
print()
|
|
return
|
|
|
|
if args.test == 'default':
|
|
test_data = [0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40]
|
|
elif args.test == 'counter':
|
|
test_data = list(range(128))
|
|
else:
|
|
test_data = [0x00, 0x01, 0x03, 0x07, 0x0F, 0x55, 0x2A]
|
|
|
|
while True:
|
|
imitator.send_sequence(test_data, args.delay)
|
|
|
|
except KeyboardInterrupt:
|
|
print("\nОстановка...")
|
|
except Exception as e:
|
|
print(f"\nОшибка: {e}")
|
|
finally:
|
|
imitator.cleanup()
|
|
print("✓ GPIO освобождены")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|