gotthard_g/aaredb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
aaredb/backend/app/data/data.py
GotthardG f588bc0cda Add endpoint for creating local contacts with access control 
Introduced a new `local_contact_router` to handle creation of local contacts. The endpoint enforces role-based access control and ensures no duplication of email addresses. Updated the router exports for consistency and cleaned up a large test file to improve readability.
2025-02-26 09:58:19 +01:00

755 lines
18 KiB
Python
Raw Blame History

from app.models import (
Contact,
Address,
Dewar,
Proposal,
Shipment,
Puck,
Sample,
DewarType,
DewarSerialNumber,
SampleEvent,
LogisticsEvent,
LocalContact,
Beamtime,
)
from datetime import datetime, timedelta
import random
import time
import hashlib
dewar_types = [
DewarType(id=1, dewar_type="Type A"),
DewarType(id=2, dewar_type="Type B"),
DewarType(id=3, dewar_type="Type C"),
]
# Define Dewar serial numbers
serial_numbers = [
DewarSerialNumber(id=1, serial_number="SN00001", dewar_type_id=1),
DewarSerialNumber(id=2, serial_number="SN00002", dewar_type_id=1),
DewarSerialNumber(id=3, serial_number="SN00003", dewar_type_id=2),
DewarSerialNumber(id=4, serial_number="SN00004", dewar_type_id=2),
DewarSerialNumber(id=5, serial_number="SN00005", dewar_type_id=3),
DewarSerialNumber(id=6, serial_number="SN00006", dewar_type_id=3),
]
# Define contact persons
contacts = [
Contact(
id=1,
pgroups="p20000, p20001",
firstname="Frodo",
lastname="Baggins",
phone_number="123-456-7890",
email="frodo.baggins@lotr.com",
),
Contact(
id=2,
pgroups="p20000, p20002",
firstname="Samwise",
lastname="Gamgee",
phone_number="987-654-3210",
email="samwise.gamgee@lotr.com",
),
Contact(
id=3,
pgroups="p20001, p20002",
firstname="Aragorn",
lastname="Elessar",
phone_number="123-333-4444",
email="aragorn.elessar@lotr.com",
),
Contact(
id=4,
pgroups="p20003, p20004",
firstname="Legolas",
lastname="Greenleaf",
phone_number="555-666-7777",
email="legolas.greenleaf@lotr.com",
),
Contact(
id=5,
pgroups="p20002, p20003",
firstname="Gimli",
lastname="Son of Gloin",
phone_number="888-999-0000",
email="gimli.sonofgloin@lotr.com",
),
Contact(
id=6,
pgroups="p20001, p20002",
firstname="Gandalf",
lastname="The Grey",
phone_number="222-333-4444",
email="gandalf.thegrey@lotr.com",
),
Contact(
id=7,
pgroups="p20000, p20004",
firstname="Boromir",
lastname="Son of Denethor",
phone_number="111-222-3333",
email="boromir.sonofdenethor@lotr.com",
),
Contact(
id=8,
pgroups="p20001, p20002",
firstname="Galadriel",
lastname="Lady of Lothlórien",
phone_number="444-555-6666",
email="galadriel.lothlorien@lotr.com",
),
Contact(
id=9,
pgroups="p20001, p20004",
firstname="Elrond",
lastname="Half-elven",
phone_number="777-888-9999",
email="elrond.halfelven@lotr.com",
),
Contact(
id=10,
pgroups="p20004, p20006",
firstname="Eowyn",
lastname="Shieldmaiden of Rohan",
phone_number="000-111-2222",
email="eowyn.rohan@lotr.com",
),
]
# Define return addresses
return_addresses = [
Address(
id=1,
pgroups="p20000, p20002",
status="active",
house_number="123",
street="Hobbiton St",
city="Hobbitbourg",
state="Shire",
zipcode="12345",
country="Middle Earth",
),
Address(
id=2,
pgroups="p20000, p20001",
status="active",
house_number="456",
street="Rohan Rd",
city="Edoras",
state="Rohan",
zipcode="67890",
country="Middle Earth",
),
Address(
id=3,
pgroups="p20001, p20002",
status="active",
house_number="789",
street="Greenwood Dr",
city="Mirkwood",
state="Greenwood",
zipcode="13579",
country="Middle Earth",
),
Address(
id=4,
pgroups="p20001, p20002, p20003",
status="active",
house_number="321",
street="Gondor Ave",
city="Minas Tirith",
state="Gondor",
zipcode="24680",
country="Middle Earth",
),
Address(
id=5,
pgroups="p20004, p20005",
status="active",
house_number="654",
street="Falgorn Pass",
city="Rivendell",
state="Rivendell",
zipcode="11223",
country="Middle Earth",
),
]
# Utilize a function to generate unique IDs
def generate_unique_id(length=16):
base_string = f"{time.time()}{random.randint(0, 10 ** 6)}"
hash_object = hashlib.sha256(base_string.encode())
hash_digest = hash_object.hexdigest()
short_unique_id = "".join(random.choices(hash_digest, k=length))
return short_unique_id
# Define dewars with unique IDs
dewars = [
Dewar(
id=1,
pgroups="p20001, p20002",
dewar_name="Dewar One",
dewar_type_id=1,
dewar_serial_number_id=2,
tracking_number="TRACK123",
return_address_id=1,
contact_id=1,
status="active",
# ready_date=datetime.strptime("2023-09-30", "%Y-%m-%d"),
# shipping_date=None,
# arrival_date=None,
# returning_date=None,
unique_id=generate_unique_id(),
),
Dewar(
id=2,
pgroups="p20001, p20002",
dewar_name="Dewar Two",
dewar_type_id=3,
dewar_serial_number_id=1,
tracking_number="TRACK124",
return_address_id=2,
contact_id=2,
status="active",
# ready_date=None,
# shipping_date=None,
# arrival_date=None,
# returning_date=None,
unique_id=generate_unique_id(),
),
Dewar(
id=3,
pgroups="p20004",
dewar_name="Dewar Three",
dewar_type_id=2,
dewar_serial_number_id=3,
tracking_number="TRACK125",
return_address_id=1,
contact_id=3,
status="active",
# ready_date=datetime.strptime("2024-01-01", "%Y-%m-%d"),
# shipping_date=None,
# arrival_date=None,
# returning_date=None,
unique_id=None,
),
Dewar(
id=4,
pgroups="p20004",
dewar_name="Dewar Four",
dewar_type_id=2,
dewar_serial_number_id=4,
tracking_number="",
return_address_id=1,
contact_id=3,
status="active",
# ready_date=datetime.strptime("2024-01-01", "%Y-%m-%d"),
# shipping_date=datetime.strptime("2024-01-02", "%Y-%m-%d"),
# arrival_date=None,
# returning_date=None,
unique_id=None,
),
Dewar(
id=5,
pgroups="p20001, p20002",
dewar_name="Dewar Five",
dewar_type_id=1,
dewar_serial_number_id=1,
tracking_number="TRACK126",
return_address_id=1,
contact_id=3,
status="active",
# arrival_date=datetime.strptime("2024-01-03", "%Y-%m-%d"),
# returning_date=datetime.strptime("2024-01-07", "%Y-%m-%d"),
unique_id=None,
),
]
logistics_events = [
LogisticsEvent(
id=1,
event_type="in preparation",
dewar_id=1,
timestamp=datetime.strptime("2024-12-03", "%Y-%m-%d"),
),
LogisticsEvent(
id=2,
event_type="ready for shipping",
dewar_id=1,
timestamp=datetime.strptime("2024-12-03", "%Y-%m-%d"),
),
LogisticsEvent(
id=3,
event_type="shipped",
dewar_id=1,
timestamp=datetime.strptime("2024-12-04", "%Y-%m-%d"),
),
LogisticsEvent(
id=4,
event_type="arrived",
dewar_id=1,
timestamp=datetime.strptime("2024-12-06", "%Y-%m-%d"),
),
LogisticsEvent(
id=5,
event_type="returned",
dewar_id=1,
timestamp=datetime.strptime("2024-12-12", "%Y-%m-%d"),
),
LogisticsEvent(
id=6,
event_type="in preparation",
dewar_id=2,
timestamp=datetime.strptime("2025-01-06", "%Y-%m-%d"),
),
LogisticsEvent(
id=7,
event_type="ready",
dewar_id=2,
timestamp=datetime.strptime("2025-01-06", "%Y-%m-%d"),
),
LogisticsEvent(
id=8,
event_type="shipped",
dewar_id=2,
timestamp=datetime.strptime("2025-01-06", "%Y-%m-%d"),
),
LogisticsEvent(
id=9,
event_type="in preparation",
dewar_id=3,
timestamp=datetime.strptime("2025-01-06", "%Y-%m-%d"),
),
LogisticsEvent(
id=10,
event_type="ready",
dewar_id=3,
timestamp=datetime.strptime("2025-01-07", "%Y-%m-%d"),
),
LogisticsEvent(
id=11,
event_type="in preparation",
dewar_id=4,
timestamp=datetime.strptime("2025-01-12", "%Y-%m-%d"),
),
LogisticsEvent(
id=12,
event_type="ready",
dewar_id=5,
timestamp=datetime.strptime("2025-01-12", "%Y-%m-%d"),
),
LogisticsEvent(
id=12,
event_type="shipped",
dewar_id=5,
timestamp=datetime.strptime("2025-01-13", "%Y-%m-%d"),
),
LogisticsEvent(
id=12,
event_type="delayed",
dewar_id=5,
timestamp=datetime.strptime("2025-01-15", "%Y-%m-%d"),
),
]
# Define proposals
proposals = [
Proposal(id=1, number="202400125"),
Proposal(id=2, number="202400235"),
Proposal(id=3, number="202400237"),
Proposal(id=4, number="202400336"),
Proposal(id=5, number="202400255"),
]
# Define shipment specific dewars
specific_dewar_ids1 = [5]
specific_dewar_ids2 = [1, 2]
specific_dewar_ids3 = [3, 4]
specific_dewars1 = [dewar for dewar in dewars if dewar.id in specific_dewar_ids1]
specific_dewars2 = [dewar for dewar in dewars if dewar.id in specific_dewar_ids2]
specific_dewars3 = [dewar for dewar in dewars if dewar.id in specific_dewar_ids3]
local_contacts = [
LocalContact(
id=1,
firstname="John",
lastname="Rambo",
phone_number="+410000000",
email="john.rambo@war.com",
),
LocalContact(
id=2,
firstname="John",
lastname="Mclane",
phone_number="+9990000099",
email="john.mclane@war.com",
),
]
beamtimes = [
Beamtime(
id=1,
pgroups="p20001",
beamtime_name="p20001-test",
beamline="X06DA",
start_date=datetime.strptime("06.02.2025", "%d.%m.%Y").date(),
end_date=datetime.strptime("07.02.2025", "%d.%m.%Y").date(),
status="confirmed",
comments="this is a test beamtime",
proposal_id=1,
local_contact_id=1,
),
Beamtime(
id=2,
pgroups="p20002",
beamtime_name="p20001-test",
beamline="X06DA",
start_date=datetime.strptime("07.02.2025", "%d.%m.%Y").date(),
end_date=datetime.strptime("08.02.2025", "%d.%m.%Y").date(),
status="confirmed",
comments="this is a test beamtime",
proposal_id=2,
local_contact_id=2,
),
]
# Define shipments
shipments = [
Shipment(
id=1,
pgroups="p20001, p20003",
shipment_date=datetime.strptime("2024-10-10", "%Y-%m-%d"),
shipment_name="Shipment from Mordor",
shipment_status="Delivered",
contact_id=2,
proposal_id=3,
return_address_id=1,
comments="Handle with care",
dewars=specific_dewars1,
),
Shipment(
id=2,
pgroups="p20001, p20002",
shipment_date=datetime.strptime("2024-10-24", "%Y-%m-%d"),
shipment_name="Shipment from Mordor",
shipment_status="In Transit",
contact_id=4,
proposal_id=4,
return_address_id=2,
comments="Contains the one ring",
dewars=specific_dewars2,
),
Shipment(
id=3,
pgroups="p20004",
shipment_date=datetime.strptime("2024-10-28", "%Y-%m-%d"),
shipment_name="Shipment from Mordor",
shipment_status="In Transit",
contact_id=5,
proposal_id=5,
return_address_id=1,
comments="Contains the one ring",
dewars=specific_dewars3,
),
]
# Define pucks
pucks = [
Puck(
id=1,
puck_name="PUCK-001",
puck_type="Unipuck",
puck_location_in_dewar=1,
dewar_id=1,
),
Puck(
id=2,
puck_name="PUCK002",
puck_type="Unipuck",
puck_location_in_dewar=2,
dewar_id=1,
),
Puck(
id=3,
puck_name="PUCK003",
puck_type="Unipuck",
puck_location_in_dewar=3,
dewar_id=1,
),
Puck(
id=4,
puck_name="PUCK004",
puck_type="Unipuck",
puck_location_in_dewar=4,
dewar_id=1,
),
Puck(
id=5,
puck_name="PUCK005",
puck_type="Unipuck",
puck_location_in_dewar=5,
dewar_id=1,
),
Puck(
id=6,
puck_name="PUCK006",
puck_type="Unipuck",
puck_location_in_dewar=6,
dewar_id=1,
),
Puck(
id=7,
puck_name="PUCK007",
puck_type="Unipuck",
puck_location_in_dewar=7,
dewar_id=1,
),
Puck(
id=8,
puck_name="PK001",
puck_type="Unipuck",
puck_location_in_dewar=1,
dewar_id=2,
),
Puck(
id=9,
puck_name="PK002",
puck_type="Unipuck",
puck_location_in_dewar=2,
dewar_id=2,
),
Puck(
id=10,
puck_name="PK003",
puck_type="Unipuck",
puck_location_in_dewar=3,
dewar_id=2,
),
Puck(
id=11,
puck_name="PK004",
puck_type="Unipuck",
puck_location_in_dewar=4,
dewar_id=2,
),
Puck(
id=12,
puck_name="PK005",
puck_type="Unipuck",
puck_location_in_dewar=5,
dewar_id=2,
),
Puck(
id=13,
puck_name="PK006",
puck_type="Unipuck",
puck_location_in_dewar=6,
dewar_id=2,
),
Puck(
id=14,
puck_name="P001",
puck_type="Unipuck",
puck_location_in_dewar=1,
dewar_id=3,
),
Puck(
id=15,
puck_name="P002",
puck_type="Unipuck",
puck_location_in_dewar=2,
dewar_id=3,
),
Puck(
id=16,
puck_name="P003",
puck_type="Unipuck",
puck_location_in_dewar=3,
dewar_id=3,
),
Puck(
id=17,
puck_name="P004",
puck_type="Unipuck",
puck_location_in_dewar=4,
dewar_id=3,
),
Puck(
id=18,
puck_name="P005",
puck_type="Unipuck",
puck_location_in_dewar=5,
dewar_id=3,
),
Puck(
id=19,
puck_name="P006",
puck_type="Unipuck",
puck_location_in_dewar=6,
dewar_id=3,
),
Puck(
id=20,
puck_name="P007",
puck_type="Unipuck",
puck_location_in_dewar=7,
dewar_id=3,
),
Puck(
id=21,
puck_name="PC002",
puck_type="Unipuck",
puck_location_in_dewar=2,
dewar_id=4,
),
Puck(
id=22,
puck_name="PC003",
puck_type="Unipuck",
puck_location_in_dewar=3,
dewar_id=4,
),
Puck(
id=23,
puck_name="PC004",
puck_type="Unipuck",
puck_location_in_dewar=4,
dewar_id=4,
),
Puck(
id=24,
puck_name="PC005",
puck_type="Unipuck",
puck_location_in_dewar=5,
dewar_id=4,
),
Puck(
id=25,
puck_name="PC006",
puck_type="Unipuck",
puck_location_in_dewar=6,
dewar_id=4,
),
Puck(
id=26,
puck_name="PC007",
puck_type="Unipuck",
puck_location_in_dewar=7,
dewar_id=4,
),
Puck(
id=27,
puck_name="PKK004",
puck_type="Unipuck",
puck_location_in_dewar=4,
dewar_id=5,
),
Puck(
id=28,
puck_name="PKK005",
puck_type="Unipuck",
puck_location_in_dewar=5,
dewar_id=5,
),
Puck(
id=29,
puck_name="PKK006",
puck_type="Unipuck",
puck_location_in_dewar=6,
dewar_id=5,
),
Puck(
id=30,
puck_name="PKK007",
puck_type="Unipuck",
puck_location_in_dewar=7,
dewar_id=5,
),
]
# Define samples
samples = []
sample_id_counter = 1
for puck in pucks:
positions_with_samples = random.randint(1, 16)
occupied_positions = random.sample(range(1, 17), positions_with_samples)
for pos in range(1, 17):
if pos in occupied_positions:
sample = Sample(
id=sample_id_counter,
sample_name=f"Sample{sample_id_counter:03}",
position=pos,
puck_id=puck.id,
)
samples.append(sample)
sample_id_counter += 1
# Define possible event types for samples
event_types = ["Mounted", "Failed", "Unmounted", "Lost"]
def generate_sample_events(samples, chance_no_event=0.2, chance_lost=0.1):
"""Generate events for samples with timestamps
increasing between different samples."""
events = []
# Set the start time to yesterday at 9:33 AM
start_time = datetime.now().replace(
hour=9, minute=33, second=0, microsecond=0
) - timedelta(days=1)
for sample in samples:
current_time = start_time
# Skip some samples with no events
if random.random() < chance_no_event:
# Increment start_time for the next sample to reflect time passage
start_time += timedelta(minutes=10)
continue
# Determine initial event type
event_type = "Failed" if random.random() < 0.05 else "Mounted"
# Append the initial event
events.append(
SampleEvent(
sample_id=sample.id, event_type=event_type, timestamp=current_time
)
)
current_time += timedelta(
seconds=50
) # Increment the time for subsequent events
# Proceed if mounted and it's not the last sample
if event_type == "Mounted" and sample is not samples[-1]:
# Determine follow-up event
if random.random() < chance_lost:
events.append(
SampleEvent(
sample_id=sample.id, event_type="Lost", timestamp=current_time
)
)
else:
events.append(
SampleEvent(
sample_id=sample.id,
event_type="Unmounted",
timestamp=current_time,
)
)
# Increment start_time for the next sample
start_time += timedelta(minutes=10)
return events
sample_events = generate_sample_events(samples)
Reference in New Issue View Git Blame Copy Permalink